1、django应用Celery
django框架请求/响应的过程是同步的，框架本身无法实现异步响应。
但是我们在项目过程中会经常会遇到一些耗时的任务, 比如：发送邮件、发送短信、大数据统计等等，这些操作耗时长，同步执行对用户体验非常不友好，那么在这种情况下就需要实现异步执行。
异步执行前端一般使用ajax，后端使用Celery。
2 、项目应用
django项目应用celery，主要有两种任务方式，一是异步任务（发布者任务），一般是web请求，二是定时任务。
celery组成请看celery介绍_宠乖仪的博客-CSDN博客
本文使用的是redis数据库作为消息中间件和结果存储数据库
1.异步任务redis
1.安装库
pip install celery
pip install redis2.celery.py
在主项目目录下，新建 celery.py 文件：
import os
import django
from celery import Celery
from django.conf import settings
# 设置系统环境变量，安装django，必须设置，否则在启动celery时会报错
# celery_study 是当前项目名
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'celery_study.settings')
django.setup()
celery_app = Celery('celery_study')
celery_app.config_from_object('django.conf:settings')
celery_app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
注意：是和settings.py文件同目录，一定不能建立在项目根目录，不然会引起 celery 这个模块名的命名冲突 ?
同时，在主项目的init.py中，添加如下代码：
from .celery import celery_app
__all__ = ['celery_app']?
3.settings.py
在配置文件中配置对应的redis配置：
# Broker配置，使用Redis作为消息中间件
BROKER_URL = 'redis://127.0.0.1:6379/0'
# BACKEND配置，这里使用redis
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'
# 任务结果过期时间，秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24
# 时区配置
CELERY_TIMEZONE='Asia/Shanghai'
# 指定导入的任务模块，可以指定多个
#CELERY_IMPORTS = (
# 'other_dir.tasks',
#)?
注意：所有配置的官方文档：Configuration and defaults — Celery 5.2.0b3 documentation ?
4.tasks.py
在子应用下建立各自对应的任务文件tasks.py(必须是tasks.py这个名字，不允许修改)
from celery import shared_task
@shared_task
def add(x, y):
return x + y
@shared_task
def mul(x, y):
return x * y
@shared_task
def xsum(numbers):
return sum(numbers)
5.调用任务
from .tasks import *
# Create your views here.
def task_add_view(request):
add.delay(100,200)
return HttpResponse(f'调用函数结果')?
?
6.启动celery
pip install eventletcelery -A celery_study worker -l debug -P eventlet注意 ：celery_study是项目名
-
使用redis时，有可能会出现如下类似的异常
AttributeError: 'str' object has no attribute 'items'这是由于版本差异，需要卸载已经安装的python环境中的 redis 库，重新指定安装特定版本（celery4.x以下适用 redis2.10.6， celery4.3以上使用redis3.2.0以上）：
xxxxxxxxxx?pip install redis==2.10.67.获取任务结果
在 views.py 中，通过 AsyncResult.get() 获取结果
from celery import result
def get_result_by_taskid(request):
task_id = request.GET.get('task_id')
# 异步执行
ar = result.AsyncResult(task_id)
if ar.ready():
return JsonResponse({'status': ar.state, 'result': ar.get()})
else:
return JsonResponse({'status': ar.state, 'result': ''})
?
AsyncResult类的常用的属性和方法：
-
state: 返回任务状态，等同status；
-
task_id: 返回任务id；
-
result: 返回任务结果，同get()方法；
-
ready(): 判断任务是否执行以及有结果，有结果为True，否则False；
-
info(): 获取任务信息，默认为结果；
-
wait(t): 等待t秒后获取结果，若任务执行完毕，则不等待直接获取结果，若任务在执行中，则wait期间一直阻塞，直到超时报错；
-
successful(): 判断任务是否成功，成功为True，否则为False；
2.定时任务
?? ?在第一步的异步任务的基础上，进行部分修改即可
1.settings.py
from celery.schedules import crontab
CELERYBEAT_SCHEDULE = {
'mul_every_30_seconds': {
# 任务路径
'task': 'celery_app.tasks.mul',
# 每30秒执行一次
'schedule': 5,
'args': (14, 5)
}
}
说明（更多内容见文档：Periodic Tasks — Celery 5.2.0b3 documentation）：
-
task：任务函数
-
schedule：执行频率，可以是整型（秒数），也可以是timedelta对象，也可以是crontab对象，也可以是自定义类（继承celery.schedules.schedule）
-
args：位置参数，列表或元组
-
kwargs：关键字参数，字典
-
options：可选参数，字典，任何 apply_async() 支持的参数
-
relative：默认是False，取相对于beat的开始时间；设置为True，则取设置的timedelta时间
在task.py中设置了日志
from celery import shared_task
import logging
logger = logging.getLogger(__name__))
@shared_task
def mul(x, y):
logger.info('___mul__'*10)
return x * y2.启动celery
(两个cmd)分别启动worker和beat
?
celery worker -A celery_study -l debug -P eventlet
celery beat -A celery_study -l debug