Django中celery的使用

一、celery的使用

celery架构图（核心）：

1.1 使用场景及安装

使用celery，主要有以下几种情况
- 定时任务
- 延迟任务
- 异步任务
使用命令安装
pip install celery

官方使用django-celery的文档：
https://docs.celeryq.dev/en/main/django/first-steps-with-django.html#using-celery-with-django

1.2在项目路径下创建celery.py

import os
from celery import Celery
# 1 导入django的配置文件(可以在WSGI中找到)---》后续在celery的任务中，就可以直接使用django的orm，缓存
os.environ.setdefault('DJANGO_SETTINGS_MODULE','TB_test.settings')# 2 实例化得到celery对象
app = Celery('myproj')# 3 celery的配置，使用django 配置文件中的配置--》刚刚写的配置
app.config_from_object('django.conf:settings')# 4 这句话会去所有app中，自动查找 tasks.py 文件，作为任务文件
app.autodiscover_tasks()

1.3 django的配置文件settings.py

############# celery的配置信息#######1 Broker配置，使用Redis作为消息中间件
BROKER_URL ='redis://127.0.0.1:6379/1'#2 BACKEND配置，使用redis
RESULT_BACKEND ='redis://127.0.0.1:6379/2'#3 序列化方案--》json
ACCEPT_CONTENT =['json']
TASK_SERIALIZER ='json'# 结果序列化方案
RESULT_SERIALIZER ='json'# 4 任务结果过期时间，秒
TASK_RESULT_EXPIRES =60*60*24#5 时区配置
TIMEZONE ='Asia/Shanghai'

1.4 在不同app中，创建tasks.py

from celery import shared_task
# 任务函数@shared_taskdefadd(a, b):return a + b

1.5 在项目目录下的 init.py中进行配置

加入以下代码

from.celery import app as celery_app
__all__ =('celery_app',)

二、异步任务

2.1 写一个视图函数测试

views.py中加入路由
urlpatterns =[...
path('user/get_mobile/',celery_test.get_mobie,name='celery_test'),...]###############视图函数中 views.py###############from django.http import JsonResponse
from Light.tasks import add
from django.shortcuts import HttpResponse
defget_mobie(request):
    mobile =int(request.GET.get('mobile'))print(mobile)# 通过delay来调用异步方法，如果在tasks中，# add这个函数没有被shared_task修饰，那么这里就会报错
    res = add.delay(mobile,5)return JsonResponse({'code':200,'msg': res.id})

2.2 启动worker

#celery -A 项目名 worker -l debug
celery -A TB_test worker -l debug

ps：好像在windows需要额外

pip install eventlet

celery -A TB_test worker worker -l debug -P eventlet

2.3 启动浏览器，输入地址，查看结果

2.4 woker获取任务，执行

三、定时任务

3.1 django的配置文件配置

from datetime import timedelta
CELERYBEAT_SCHEDULE ={'every_5_second':{'task':'Light.tasks.add',# 设置每隔5s执行一次add任务'schedule': timedelta(seconds=5),'args':(33,44),}}

3.2 重启woker，启动beat

celery执行定时任务的是beat，在开头的celery框架可以看出

# 启动一个worker
celery -A TB_test worker -l debug
# 启动定时任务
celery -A TB_test beat -l debug

beat就会按照配置时间5s把任务提交一次

四、通过Admin 后台管理添加定时任务

CELERYBEAT_SCHEDULE ={'every_5_second':{'task':'Light.tasks.add',# 设置每隔5s执行一次add任务'schedule': timedelta(seconds=5),'args':(33,44),}}# 现目前存在问题：
    如果要再新加一个定时任务（例如爬新闻数据）
    我们只能修改代码 ：settings.py-->加入代码
    重启worker，重启beat--》才能行
    这样操作起来很麻烦
    
# 所以想要直接通过点点点，就能自动添加定时任务
    那么就可以引入 
    - Admin 后台管理添加定时任务

4.1 引入–安装djiango-celery-beat

pip install django-celery-beat

4.2 在app中注册djiango-celery-beat

INSTALLED_APPS =[...'app01','django_celery_beat'# 注册刚刚第三方模块]

4.3 配置时区

# settings.py
LANGUAGE_CODE ='zh-hans'
TIME_ZONE ='Asia/Shanghai'
USE_I18N =True
USE_L10N =True
USE_TZ =False# 配置调度器# 只要配了这个，原来celery中的定时任务统一不能用了，需要我们手动配置了
CELERYBEAT_SCHEDULER ='django_celery_beat.schedulers.DatabaseScheduler'

4.4 配置数据库，进行数据迁移

- 需要先安装mysql，用mysql创建一个数据库
- 再安装python中链接mysql需要用到的
- pip install mysqlclient
- pip install pymysql

################ settings.py ################
DATABASES ={'default':{'ENGINE':'django.db.backends.mysql','NAME':'Simpleuse',# 数据库名字'USER':'root','PASSWORD':'xxx','HOST':'127.0.0.1',# ip'PORT':3306,}}################  app01/__init__.py ################import pymysql
pymysql.install_as_MySQLdb()

命令行执行
python3 manage.py makemigrations
python3 manage.py migrate
#  django-celery-beat 这个app会产生表--》以后只要咱么在表里添加记录，就能制定定时任务

到此，数据库表就创建好了。

4.5 使用admin后台管理插入数据

4.5.1 创建账号

-django提供了一个后台管理--》登录到后台管理，可以图形化界面录入数据
    -但是需要个账号登录:命令创建账号
        python3 manage.py createsuperuser

4.5.2 打开网页，登录

界面略微丑陋，美化一下

4.6 美化admin界面

# 1  simpleui 美化admin# 2 开源地址
https://gitee.com/tompeppa/simpleui
# 3 文档地址
https://newpanjing.github.io/simpleui_docs/config.html
    
# 4 安装
pip3 install django-simpleui
# 5 app中注册
INSTALLED_APPS =['simpleui',...]

4.7 手动添加周期任务

# 1 启动worker，beat# 启动worker
celery -A Simpleuse worker -l debug
# 启动beat
celery -A Simpleuse beat -l debug

五、通过Admin查看任务运行情况

# 在控制台监控任务执行情况或者去redis中查看，还不是很方便，最好是能够通过web界面看到任务的执行情况，如有多少任务在执行，有多少任务执行失败了等。# 这个Celery也是可以做到了，就是将任务执行结果写到数据库中，通过web界面显示出来。
这里要用到django-celery-results插件。
通过插件可以使用Django的orm作为结果存储，这样的好处在于我们可以直接通过django的数据查看到任务状态，同时为可以制定更多的操作

5.1 安装django-celery-results

pip install django-celery-results

5.2 在app中注册

INSTALLED_APPS =(...,'django_celery_results',)

5.3 修改配置文件

# 让之前backend----》结果存储放在redis中----》改到放到数据中---》因为在数据库中的咱们可以通过admin查看#RESULT_BACKEND = 'redis://127.0.0.1:6379/1'# 使用使用django orm 作为结果存储
CELERY_RESULT_BACKEND ='django-db'#使用django orm 作为结果存储

5.4 迁移数据库

# 执行迁移命令 
python3 manage.py makemigrations
python3 manage.py migrate

可以看到数据库多出了3张存储结果的表

六、通过flower插件监控任务情况

Flower 是一个用于监控和管理 Celery 集群的开源 Web 应用程序。它提供有关 Celery  workers 和tasks状态的实时信息
# Flower可以：1 实时监控celery的Events
    -查看任务进度和历史记录
    -查看任务详细信息（参数、开始时间、运行时间等）
2 远程操作
    -查看workers 状态和统计数据
    -关闭并重新启动workers 实例
    -控制工作池大小和自动缩放设置
    -查看和修改工作实例消耗的队列
    -查看当前正在运行的任务
    -查看计划任务（预计到达时间/倒计时）
    -查看保留和撤销的任务
    -应用时间和速率限制
    -撤销或终止任务
3 Broker 监控
    -查看所有 Celery 队列的统计信息

6.1 安装flower和启动

#1  安装
pip install flower
# 2 启动# celery -A 项目名 flower --port-指定端口号# 项目名变了，这里下班了，在家换了一个电脑写，但是逻辑是一样的。
celery -A Simpleuse flower --port-5555#3 浏览器访问：
http://127.0.0.1:5555/
获取更详细：http://127.0.0.1:5555/tasks
        
# 4 启动 worker，beat，flower-一般先启动flower

这个要比admin的看起来更友好一些