这些情况中的示例如下:
- 周期性任务 —— 计划在特定时间间隔内运行的工作。例如,每日、每月的报告生成。
- 第三方工具 —— 应用程序应该快速向用户返回响应,而不是等待其他任务先完成。例如,发送电子邮件、通知,将更新进度传递给内部工具。
- 长时间运行的工作 —— 执行复杂或资源昂贵的工作,并且用户需要等待工作完成。例如。DAG工作流、基于Map-Reduce的任务、长时间运行的Spark作业等。
那么,如何处理这些情况呢?这时,Celery就派上用场了。
什么是Celery?
Celery是一个开源的任务队列实现,通常与基于Python的网络框架(如Flask和Django)相结合,在典型的请求-响应周期之外异步执行任务。
因此,Celery本质上是一个基于分布式消息传递的任务队列。执行单元或任务在一个或多个worker上使用多处理、gevent或Eventlet同时执行。这些任务可以同步执行(即等到准备就绪)或异步执行(即在后台)。
Celery是如何工作的?
Celery是一个分布式任务队列,基于生产者-消费者模式。
任务队列是用于跨线程和机器分配工作的机制,本质上是生产者(Web应用程序)和消费者(Celery工作者)之间的消息中介。
Celery通过消息进行交互,代理(broker)在客户(生产者)和工作者(消费者)之间充当中间人。为了启动任务,客户端将消息推送到队列中,然后代理将该消息传递给工作者。
Celery系统可以由多个worker和broker组成,这为高可用性和横向扩展提供了可能。
简而言之,Celery客户端是生产者,它通过消息代理向队列中添加新的任务。然后,Celery工作者同样通过消息代理从队列中获取新的任务。一旦处理完毕,结果就会存储在结果后端。
工作实例
下面的例子将使用RedisMQ作为消息代理。
设置Redis
在linux/macOS系统上,通过以下命令在本地运行Redis服务器:
设置好Redis后,通过执行以下命令运行Redis服务器:
该服务器在默认的6379端口运行。
设置应用程序
首先,在本地设置Python项目。
Celery可以通过标准工具如pip或easy_install来安装。通过以下命令安装Celery和Redis:
现在需要一个Celery实例来运行应用程序,Celery实现任何任务都是以实例开始,比如创建和管理任务等。
在项目中创建一个文件tasks.py:
这里定义了一个简单的任务add(),返回两个数字的总和。
运行Celery Worker
在终端上,切换到项目位置并用以下命令运行Celery worker:
关于Celery worker命令行的详细信息,可以使用help:
调用任务
在Celery中,使用delay()方法来调用任务。
打开项目的另一个终端窗口并运行以下命令:
这将打开Python命令行。
这将返回一个AsyncResult实例,可以用来检查任务状态,获得其返回值,等待任务完成,也可以在失败时获得异常和回溯。
运行add.delay()命令后,任务会被推送到队列中,然后被worker获取。这可以在Celery worker终端上进行验证,可以清楚地看到任务被接收,之后任务成功完成。