Django Celery异步任务队列的实现
背景
在开发中，我们常常会遇到一些耗时任务，举个例子：
上传并解析一个1w条数据的Excel文件，最后持久化至数据库。
在我的程序中，这个任务耗时大约6s，对于用户来说，6s的等待已经是个灾难了。
比较好的处理方式是：
- 接收这个任务的请求
- 将这个任务添加到队列中
- 立即返回「操作成功，正在后台处理」的字样
- 后台消费这个队列，执行这个任务
我们按照这个思路，借助Celery进行实现。
实现
本文所使用的环境如下：
- Python3.6.7
- RabbitMQ3.8
- Celery4.3
使用Docker安装RabbitMQ
Celery依赖一个消息后端，可选方案有RabbitMQ,Redis等，本文选用RabbitMQ。
同时为了安装方便，RabbitMQ我直接使用Docker安装：
dockerrun-d--nameanno-rabbit-p5672:5672rabbitmq:3
启动成功后，即可通过amqp://localhost访问该消息队列。
安装并配置Celery
Celery是Python实现的工具，安装可以直接通过Pip完成：
pipinstallcelery
同时假设当前我的项目文件夹为proj，项目名为myproj，应用名为myapp
安装完成后，在proj/myproj/路径下创建一个celery.py文件，用来初始化Celery实例：
proj/myproj/celery.py
from__future__importabsolute_import,unicode_literals
importos
fromceleryimportCelery,platforms
#setthedefaultDjangosettingsmoduleforthe'celery'program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE','myproj.settings')
app=Celery('myproj',
broker='amqp://localhost//',
backend='amqp://localhost//')
#Usingastringheremeanstheworkerdon'thavetoserialize
#theconfigurationobjecttochildprocesses.s
#-namespace='CELERY'meansallcelery-relatedconfigurationkeys
#shouldhavea`CELERY_`prefix.
app.config_from_object('django.conf:settings',namespace='CELERY')
#LoadtaskmodulesfromallregisteredDjangoappconfigs.
app.autodiscover_tasks()
然后在proj/myproj/__init__.py中添加对Celery对象的引用，确保Django启动后能够初始化Celery：
proj/myproj/__init__.py
from__future__importabsolute_import,unicode_literals
#Thiswillmakesuretheappisalwaysimportedwhen
#Djangostartssothatshared_taskwillusethisapp.
from.celeryimportappascelery_app
__all__=('celery_app',)
无其他特殊配置的话，Celery的基本配置就是这些。
编写一个耗时任务
为了模拟一个耗时任务，我们直接创建一个方法，使其「睡」10s，并将其设置为Celery的任务：
proj/myapp/tasks.py
importtime frommyproj.celeryimportappascelery_app @celery_app.task defwaste_time(): time.sleep(10) return"Runfunction'waste_time'finished."
启动CeleryWorker
Celery配置完成，并且任务创建成功后，我们以异步任务的模式启动Celery：
celery-Amyprojworker-linfo
注意到我强调了异步模式，是因为Celery除了支持异步任务，还支持定时任务，因此启动时候要指明。
同时要注意，Celery一旦启动，对Task(此处为waste_time)的修改必须重启Celery才会生效。
任务调用
在请求处理的逻辑代码中，调用上面创建好的任务：
proj/myapp/views.py
fromdjango.httpimportJsonResponse
fromdjango.views.decorators.httpimportrequire_http_methods
from.tasksimportwaste_time
@require_http_methods(["POST"])
defupload_files(request):
waste_time.delay()
#Statuscode202:Accepted,表示异步任务已接受，可能还在处理中
returnJsonResponse({"results":"操作成功，正在上传，请稍候..."},status=202)
调用waste_time.delay()方法后，waste_time会被加入到任务队列中，等待空闲的CeleryWorker调用。
效果
当我们发送请求时，这个接口会直接返回{"results":"操作成功，正在上传，请稍候..."}的响应内容而非卡住十秒，用户体验要好许多。
总结
用Celery处理这种异步任务是Python常用的方法，虽然实际执行成功耗时不变甚至有所增加（如Worker繁忙导致处理滞后），但是对于用户体验来说更容易接受，点击上传大文件后可以继续处理其他事务，而不需要在页面等待。
Celery还有更多用法本文未介绍到，其文档已经非常详尽，有需要可直接参考。
参考
http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html
https://hub.docker.com/_/rabbitmq
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。