celery是python的一个第三方库,中文为"芹菜",是一个基于生产者消费者模式的框架,主要用于异步执行任务或定时任务。本文将介绍如何使用celery实现异步执行任务的方法。
一. 安装Celery和后端代理
首先,需要安装Celery,并安装RabbitMQ或Redis作为后端代理。
# 安装Celery pip install celery
RabbitMQ和Redis都可以作为Celery的后端代理,用于任务队列。Celery官方文档建议优先使用RabbitMQ,但也可以选择Redis。安装RabbitMQ需要依赖Erlang,具体的安装教程可以自行查找。
二. 搭建Celery任务架构
在项目中合适的位置创建一个名为celery_tasks的目录,在这个目录下编写Celery的相关代码,将Celery代码与项目业务逻辑代码分离。注意:目录名不要直接使用celery,以免与Python关键字或第三方模块冲突,导致导包错误。
在celery_tasks目录下创建config.py, tasks.py, main.py三个Python文件,分别用于编写Celery的配置代码、任务函数代码和任务启动代码。
# 目录结构
- celery_tasks
- config.py
- main.py
- tasks.py三. 编写代码实现异步调用任务
config.py
from celery import Celery
<h1>创建Celery对象, 'demo'是自定义的命名</h1><h1>broker指定后端代理,可以使用RabbitMQ或Redis</h1><p>app = Celery('demo', broker='amqp://guest@localhost:5672//')</p><h1>app = Celery('demo', broker='redis://127.0.0.1:6379/15')tasks.py
from config import app</h1><h1>定义任务,使用@app.task装饰器,Celery会自动识别任务</h1><p>@app.task(name='celery_task1_name')
def celery_task1_name(arg):
print('执行任务代码', arg)</p><p>@app.task(name='celery_task2_name')
def celery_task2_name():
print('将需要执行的代码导入tasks.py文件,然后在这里调用即可')main.py
from tasks import *</p><h1>设置Celery对象自动识别任务</h1><h1>'celery_tasks'指定tasks.py的目录,保证程序能找到tasks.py</h1><p>app.autodiscover_tasks(['celery_tasks'])
四. 启动Celery任务
在main.py所在目录下执行以下命令启动Celery任务。如果不在此目录,则需要在main前加上相对路径,例如:celery_tasks.main。
celery -A main worker -l info
参数说明:
-
-A指定Celery的启动入口main -
worker为Celery执行任务的后端工人 -
-l指定日志级别为info
执行成功后,Celery会启动worker,从代理队列中获取任务并执行。如果任务队列为空,则一直等待直到有任务。
Windows Bug:
如果使用Celery4.0以上版本在Windows上,通过上述命令启动,在执行task.delay()时会报错:ValueError: not enough values to unpack (expected 3, got 0)。此问题在Linux上不会出现,只在Windows上存在,与“绿色线程”有关,具体可参考eventlet相关资料。
解决办法:
- 安装eventlet
pip install eventlet
- 启动worker时增加
-P eventlet参数
celery -A main worker -l info -P eventlet
五. 调用Celery异步执行任务
在需要执行异步任务的地方导入任务,使用task.delay(参数)调用任务。例如,在与celery_tasks目录同级的demo目录下的demo.py文件中异步执行任务。
from celery_tasks.tasks import celery_task1_name, celery_task2_name</p><p>def demo_func(a):</p><h1>调用格式:任务名.delay(参数)</h1><pre class="brush:php;toolbar:false;">celery_task1_name.delay(a)
print('celery_task1_name执行完成:{}!'.format(a))
celery_task2_name.delay()
print('celery_task2_name执行完成!')demo_func('hello celery!')
至此,我们已经实现了Celery异步调用任务的功能,通过复制以上步骤中的代码,可以实现异步任务的demo。
