Django Celery异步任务队列的实现

背景

在开发中,我们常常会遇到一些耗时任务,举个例子:

上传并解析一个 1w 条数据的 Excel 文件,最后持久化至数据库。

在我的程序中,这个任务耗时大约 6s,对于用户来说,6s 的等待已经是个灾难了。

比较好的处理方式是:

  1. 接收这个任务的请求
  2. 将这个任务添加到队列中
  3. 立即返回「操作成功,正在后台处理」的字样
  4. 后台消费这个队列,执行这个任务

我们按照这个思路,借助 Celery 进行实现。

实现

本文所使用的环境如下:

  • Python 3.6.7
  • RabbitMQ 3.8
  • Celery 4.3

使用 Docker 安装 RabbitMQ

Celery 依赖一个消息后端,可选方案有 RabbitMQ, Redis 等,本文选用 RabbitMQ 。

同时为了安装方便,RabbitMQ 我直接使用 Docker 安装:

docker run -d --name anno-rabbit -p 5672:5672 rabbitmq:3

启动成功后,即可通过 amqp://localhost 访问该消息队列。

安装并配置 Celery

Celery 是 Python 实现的工具,安装可以直接通过 Pip 完成:

pip install celery

同时假设当前我的项目文件夹为 proj ,项目名为 myproj ,应用名为 myapp

安装完成后,在 proj/myproj/ 路径下创建一个 celery.py 文件,用来初始化 Celery 实例:

proj/myproj/celery.py

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery, platforms

# set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproj.settings')

app = Celery('myproj',
       broker='amqp://localhost//',
       backend='amqp://localhost//')

# Using a string here means the worker don't have to serialize
# the configuration object to child processes.s
# - namespace='CELERY' means all celery-related configuration keys
#  should have a `CELERY_` prefix.
app.config_from_object('django.conf:settings', namespace='CELERY')

# Load task modules from all registered Django app configs.
app.autodiscover_tasks()

然后在 proj/myproj/__init__.py 中添加对 Celery 对象的引用,确保 Django 启动后能够初始化 Celery:

proj/myproj/__init__.py

from __future__ import absolute_import, unicode_literals

# This will make sure the app is always imported when
# Django starts so that shared_task will use this app.
from .celery import app as celery_app

__all__ = ('celery_app',)

无其他特殊配置的话,Celery 的基本配置就是这些。

编写一个耗时任务

为了模拟一个耗时任务,我们直接创建一个方法,使其「睡」10s ,并将其设置为 Celery 的任务:

proj/myapp/tasks.py

import time
from myproj.celery import app as celery_app

@celery_app.task
def waste_time():
  time.sleep(10)
  return "Run function 'waste_time' finished."

启动 Celery Worker

Celery 配置完成,并且任务创建成功后,我们以异步任务的模式启动 Celery :

celery -A myproj worker -l info

注意到我强调了异步模式,是因为 Celery 除了支持异步任务,还支持定时任务,因此启动时候要指明。

同时要注意,Celery 一旦启动,对 Task(此处为 waste_time) 的修改必须重启 Celery 才会生效。

任务调用

在请求处理的逻辑代码中,调用上面创建好的任务:

proj/myapp/views.py

from django.http import JsonResponse
from django.views.decorators.http import require_http_methods
from .tasks import waste_time

@require_http_methods(["POST"])
def upload_files(request):
  waste_time.delay()
  # Status code 202: Accepted, 表示异步任务已接受,可能还在处理中
  return JsonResponse({"results": "操作成功,正在上传,请稍候..."}, status=202)

调用 waste_time.delay() 方法后, waste_time 会被加入到任务队列中,等待空闲的 Celery Worker 调用。

效果

当我们发送请求时,这个接口会直接返回 {"results": "操作成功,正在上传,请稍候..."} 的响应内容而非卡住十秒,用户体验要好许多。

总结

用 Celery 处理这种异步任务是 Python 常用的方法,虽然实际执行成功耗时不变甚至有所增加(如 Worker 繁忙导致处理滞后),但是对于用户体验来说更容易接受,点击上传大文件后可以继续处理其他事务,而不需要在页面等待。
Celery 还有更多用法本文未介绍到,其文档已经非常详尽,有需要可直接参考。

参考

http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html

https://hub.docker.com/_/rabbitmq

以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持我们。

时间: 2019-07-24

异步任务队列Celery在Django中的使用方法

前段时间在Django Web平台开发中,碰到一些请求执行的任务时间较长(几分钟),为了加快用户的响应时间,因此决定采用异步任务的方式在后台执行这些任务.在同事的指引下接触了Celery这个异步任务队列框架,鉴于网上关于Celery和Django结合的文档较少,大部分也只是粗粗介绍了大概的流程,在实践过程中还是遇到了不少坑,希望记录下来帮助有需要的朋友. 一.Django中的异步请求 Django Web中从一个http请求发起,到获得响应返回html页面的流程大致如下:http请求发起 --

Django使用Celery异步任务队列的使用

1 Celery简介 Celery是异步任务队列,可以独立于主进程运行,在主进程退出后,也不影响队列中的任务执行. 任务执行异常退出,重新启动后,会继续执行队列中的其他任务,同时可以缓存停止期间接收的工作任务,这个功能依赖于消息队列(MQ.Redis). 1.1 Celery原理 Celery的 架构 由三部分组成,消息中间件(message broker),任务执行单元(worker)和任务执行结果存储(task result store)组成. 消息中间件:Celery本身不提供消息服务,但

Python Celery异步任务队列使用方法解析

Celery是一个异步的任务队列(也叫做分布式任务队列),一个简单,灵活,可靠的分布式系统,用于处理大量消息,同时为操作提供维护此类系统所需要的工具. celery的优点 1:简单,容易使用,不需要配置文件 2:高可用,任务执行失败或执行过程中发生连续中断,celery会自动尝试重新执行任务 3:快速,一个单进程的celery每分钟可以处理上百万个任务 4:灵活,几乎celery的各个组件都可以被扩展 celery应用场景 1:异步发邮件,一般发邮件等比较耗时的操作,这个时候需要提交任务给cel

详解配置Django的Celery异步之路踩坑

人生苦短,我用python. 看到这句话的时候,感觉可能确实是很深得人心,不过每每想学学,就又止步,年纪大了,感觉学什么东西都很慢,很难,精神啊注意力啊思维啊都跟不上.今天奶牛来分享自己今天踩的一个坑. 先说说配置过程吧,初学Django,啥都不懂,当然,python也很水,啥东西都得现查现用.Django安装还是很简单的. apt-get install python3 pip3 install django 嗯,就是两条命令的事儿. 再说celery的安装: pip3 install cel

Python环境下安装使用异步任务队列包Celery的基础教程

1.简介 celery(芹菜)是一个异步任务队列/基于分布式消息传递的作业队列.它侧重于实时操作,但对调度支持也很好. celery用于生产系统每天处理数以百万计的任务. celery是用Python编写的,但该协议可以在任何语言实现.它也可以与其他语言通过webhooks实现. 建议的消息代理RabbitMQ的,但提供有限支持Redis, Beanstalk, MongoDB, CouchDB, ,和数据库(使用SQLAlchemy的或Django的 ORM) . celery是易于集成Dja

Django集成celery发送异步邮件实例

安装依赖 pip install django-celery-beat pip install django-celery-email pip install celery pip install msgpack-python pip install msgpack 在settings文件中配置 添加app应用到installed_apps中 "djcelery_email", "django_celery_beat" 修改.env文件配置: #邮箱后端,使用cel

详解django+django-celery+celery的整合实战

本篇文章主要是由于计划使用django写一个计划任务出来,可以定时的轮换值班人员名称或者定时执行脚本等功能,百度无数坑之后,终于可以凑合把这套东西部署上.本人英文不好,英文好或者希望深入学习或使用的人,建议去参考官方文档,而且本篇的记录不一定正确,仅仅实现crontab 的功能而已. 希望深入学习的人可以参考 http://docs.jinkan.org/docs/celery/ . 首先简单介绍一下,Celery 是一个强大的分布式任务队列,它可以让任务的执行完全脱离主程序,甚至可以被分配到其

python celery分布式任务队列的使用详解

一.Celery介绍和基本使用 Celery 是一个 基于python开发的分布式异步消息任务队列,通过它可以轻松的实现任务的异步处理, 如果你的业务场景中需要用到异步任务,就可以考虑使用celery, 举几个实例场景中可用的例子: 你想对100台机器执行一条批量命令,可能会花很长时间 ,但你不想让你的程序等着结果返回,而是给你返回 一个任务ID,你过一段时间只需要拿着这个任务id就可以拿到任务执行结果, 在任务执行ing进行时,你可以继续做其它的事情. 你想做一个定时任务,比如每天检测一下你们

Django中celery执行任务结果的保存方法

如下所示: pip3 install django-celery-results INSTALLED_APPS = ( ..., 'django_celery_results',) # 注意这个是下划线'_' python3 manage.py migrate django_celery_results CELERY_RESULT_BACKEND = 'django-db' #在settings.py文件中配置 注意异步任务views.py中调用时,想要记录结果必须是"任务函数.delay(*a