详解Django中异步任务之django-celery
本文详细介绍如何在Django项目中集成Celery以实现异步和定时任务。涵盖Celery架构模块、环境配置（Django、Redis版本建议）、Broker与Worker设置，以及通过django-celery创建和调用异步任务的具体步骤。
Celery官方文档参考链接：http://docs.jinkan.org/docs/celery/
相关参考文章：https://www.jb51.net/article/158046.htm
Django集成Celery概述
Django异步任务处理方案：django-celery
Celery简介：
Celery的典型应用场景：
- 处理耗时较长的任务或执行定时任务。
- 对请求返回结果不敏感的场景。
- 例如发送短信验证码，可以先向用户反馈任务状态（成功或失败），而无需等待最终结果。
- 若请求结果至关重要（如支付流程），建议直接在Django中同步处理。
Celery架构与模块
简要介绍，Celery是一个功能强大的分布式任务队列系统，它允许任务执行完全独立于主程序，甚至可分布到其他主机运行。通常用于实现异步任务（asynctask）和定时任务（crontab）。其架构组成如下图所示：
Celery主要包含以下模块：
任务模块（Task）
包含异步任务和定时任务。异步任务通常在业务逻辑中被触发并发送至任务队列，而定时任务由Celery Beat进程周期性地将任务发送至队列。
消息中间件（Broker）
Broker即任务调度队列，负责接收生产者发送的消息（任务）并存入队列。Celery本身不提供队列服务，官方推荐使用RabbitMQ或Redis等。
任务执行单元（Worker）
Worker是执行任务的处理单元，它实时监控消息队列，获取并执行调度的任务。
任务结果存储（Backend）
Backend用于存储任务的执行结果，以便查询。与消息中间件类似，存储后端也可使用RabbitMQ、Redis或MongoDB等。
django-celery环境配置
django-celery集成
首先需要统一使用环境，因为如果Redis版本过高可能会报错。
解决方法：建议降低Redis版本。
推荐版本配置如下：
Django == 2.2.6
django-celery == 3.3.1
django-redis == 4.11.0
redis == 2.10.6
celery == 3.1.26.post2
依赖安装：pip install .....
人都知道
Django配置与数据表创建
1. 修改setting.py Django配置文件，增加如下配置：
import djcelery ###导包
djcelery.setup_loader() ###
BROKER_URL = 'redis://127.0.0.1:6379/2'
# BROKER_URL='redis://192.168.217.77:16379/2' #任何可用的redis都可以，不一定要在django server运行的主机上
CELERYBEAT_SCHEDULER = 'djcelery.schedulers.DatabaseScheduler' ###
INSTALLED_APPS = [
...
"djcelery", # 加入djcelery应用
...
]
CELERY_TIMEZONE='Asia/Shanghai' #并没有北京时区，与下面TIME_ZONE应该一致
TIME_ZONE='Asia/Shanghai' #在文件开头增加上述配置，根据实际环境配置Redis的地址和端口。时区必须设置为Asia/Shanghai，否则时间不准确会影响定时任务的运行。
上述代码首先导出djcelery模块，并调用setup_loader方法加载相关配置；注意配置时区，否则默认使用UTC时间会比东八区慢8个小时。其中INSTALLED_APPS末尾添加两项，分别表示添加celery服务和自己定义的apps服务。
2. 创建Celery所需的数据表
python manage.py migrate
#如若不成功可以尝试一下命令语句
#python manage.py syncdb创建与注册异步任务
3. 创建task
stasks.py
# -*- coding: utf-8 -*-
import json, time
from syl.settings import ALY_ACCESSKEY_ID, ALY_ACCESSKEY_SECRET
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest
from celery import task
# 阿里云短信验证码
@task
def Celery_Send_Sms(phone, data):
client = AcsClient(ALY_ACCESSKEY_ID, ALY_ACCESSKEY_SECRET, 'cn-hangzhou')
request = CommonRequest()
request.set_accept_format('json')
request.set_domain('dysmsapi.aliyuncs.com')
request.set_method('POST')
request.set_protocol_type('https') # https | http
request.set_version('2017-05-25')
request.set_action_name('SendSms')
request.add_query_param('RegionId', "cn-hangzhou")
request.add_query_param('PhoneNumbers', phone)
request.add_query_param('SignName', "美多商城")
request.add_query_param('TemplateCode', "SMS_205397849")
request.add_query_param('TemplateParam', data)
response = client.do_action(request)
time.sleep(10)
print(str(response, encoding='utf-8'))
res = json.loads(str(response, encoding='utf-8'))
#celery运行命令
# python manage.py celery worker --loglevel=info- 当settings.py中的djcelery.setup_loader()运行时，Celery便会查看所有INSTALLED_APPS中app目录下的tasks.py文件，找到标记为task的function，并将它们注册为celery task。
- 在执行djcelery.setup_loader()时，task是以INSTALLED_APPS中的app名，加.tasks.function_name注册的
- 一次需要注意 在impprt task时, 需要保持一致
- 如果我们由于python path不同而使用不同的引用方式时(例如在tasks.py中使用from myproject.myapp.tasks import add形式), Celery将无法得知这是同一task, 因此可能会引起奇怪的bug。
异步任务调用与服务启动
让任务变成异步
例如我们希望在用户发出request后异步执行该task, 马上返回response, 从而不阻塞该request, 使用户有一个流畅的访问过程. 那么, 我们可以使用.delay。
views.py
import re
import random
from rest_framework.permissions import AllowAny
from django_redis import get_redis_connection
from rest_framework.views import APIView
from rest_framework.response import Response
# from utils.MyBaseView import send_message, Send_Sms
from verifications.stasks import Celery_Send_Sms
# 用户注册短信验证码
class SmsCodeView(APIView):
'''使用apiview的限流'''
# 1. 所有人可以访问
permission_classes = (AllowAny,)
def post(self, request):
# 1. 获取参数
phone = request.data.get('phone') # 手机号
image_code = request.data.get('image_code') # 字符串验证码
image_code_uuid = request.data.get('image_code_uuid') # 前端生成的uuid，是redis中图片验证码的key
# 2. 检查参数
if not all([phone, image_code, image_code_uuid]):
return Response({'code': 400, 'msg': '参数不全'})
# 检查手机号是否正确
if not re.match(r'^1[3456789]\d{9}$', phone):
return Response({"code": 999, "msg": "手机号码不正确"})
# 3. 检查是否发送
redis_client = get_redis_connection('img_code') # 连接redis数据库
# phone_exists = redis_client.get(phone)
# if phone_exists:
# return Response({"code": 999, "msg": "频繁发送, 请稍后再试"})
# 4.检查图片验证码是否合法
redis_image_code = redis_client.get(image_code_uuid) # 字符串验证码
if redis_image_code:
# bytes 转成 string
redis_image_code = redis_image_code.decode() # 把uuid解码
# 比较用户提供的图片内容是否和redis中保存的一致
if image_code.upper() != redis_image_code:
return Response({'code': 999, 'msg': '图片验证码不正确'})
# 5. 发送
code = '%06d' % random.randint(100000, 999999) # 随机6位验证码
print('code===============================================', code)
# 使用容联云短信验证码
# send_resp = send_message(phone, (code, "5"))
# 使用阿里云短信验证码
data = {'code': code}
# send_resp = Send_Sms(phone, data)
# 使用celery异步发送短信
Celery_Send_Sms.delay(phone, data) #delay是注册为celery异步任务的关键点
# Celery_Send_Sms(phone, data) # delay是注册为celery异步任务的关键点
# 5.1 保存code 到 redis中
redis_client.setex(phone, 60 * 5, code) # phone:code, 5分钟有效期
# 5.2 从redis中删除这个图片验证码, 以防再次被使用
redis_client.delete(image_code_uuid)
# 6.存储这个已经发送验证码的手机号，防止频繁发送（使用pipeline 批量操作 ）
pl = redis_client.pipeline()
pl.setex(phone, 60 * 5, code)
pl.delete(image_code_uuid)
pl.execute()
# 7. 返回结果
return Response({"code": 200, "msg": "短信发送成功"})1. 启动celery
首先正常启动你的django任务，然后启动celery服务即可。
python manage.py celery worker --loglevel=info
出现上图这个报错不让超级管理员来启动，在settings.py加入以下配置
from celery import Celery, platforms
platforms.C_FORCE_ROOT = True2. 验证celery任务
在搞定上面的东西以后，你就可以通过postman来请求接口让接口使用celery来异步执行任务而不阻塞你的request请求。
到此这篇关于详解Django中异步任务之django-celery的文章就介绍到这了,更多相关Django异步任务内容请搜索本站以前的文章或继续浏览下面的相关文章希望大家以后多多支持本站！
本站声明
本文内容仅供参考，不代表本站观点。转载或引用请注明出处。如发现内容问题，请通过联系我们页面反馈。
