阶段三-04 celery基础模块安装和试运行

版本兼容性

Django==4.0.5 #4.1.x应该也兼容 celery==5.2.6 django-celery-beat==2.3.0 django-celery-results==2.3.1 flower==1.0.0 #应该可以用1.1.0或1.2.0，自己尝试

其他配置

1、settings同级目录下创建celery.py

__init__中加入   __all__ = ('celery_app',)

import osfrom celery import Celery# Set the default Django settings module for the 'celery' program.
os.environ.setdefault('DJANGO_SETTINGS_MODULE', '项目包名（settings所在的包）')app = Celery('djangoProject1', broker='amqp://用户名:密码@IP')# Using a string here means the worker doesn't have to serialize
# the configuration object to child processes.
# - 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 apps.
app.autodiscover_tasks()@app.task(bind=True)
def debug_task(self):print(f'Request: {self.request!r}')

2、settings中

INSTALLED_APPS中加入 'django_celery_results',最下方加入：

# ################################## CELERY ######################################
# 最重要的配置，消息broker的连接方式，格式为： db://user:password@host:port/dbname
CELERY_BROKER_URL = 'amqp://xxxx/test_plt'# 设置时区，与django TIME_ZONE一致
CELERY_TIMEZONE = TIME_ZONE# 支持数据库 django-db 和缓存 django-cache
CELERY_RESULT_BACKEND = 'django-db'# 储存结果过期，默认1天，0为永不过期
# 如果beat开启，celery每天会自动清除过期记录
CELERY_RESULT_EXPIRES = 259200# 超时终止，执行下个任务
CELERY_TASK_TIME_LIMIT = 600

3、尝试启动

windows下安装插件eventlet

celery -A djangoProject1 worker -P eventlet -l info --pool=solo

命令行运行debug_task

import os, django
os.environ['DJANGO_SETTINGS_MODULE'] ='djangoProject1'
django.setup()from djangoProject1.celery import debug_taskdebug_task.delay()

返回terminal查看，有可能报错多线程执行错误（需要停掉重新 加上 --pool=solo）；如果没有报错执行成功的效果：

此时再来到jango页面点击，页面可能会报错。这个是涉及到数据库写入时间的时区冲突问题 

4、单独创建tasks.py，用于验证beat和flower

在test_plt下创建tasks.py， 引入 @shared_task ；为了让celery识别到我们的任务。注意：完成后需要重启woker才能识别到新的task

成功执行后，Result Data会记录task的返回值

执行后需要监控消息队列，发现rabbit get message （ack mode中 Nack…… 表示提取消息后再重新入列，以免celery worker 消费不到）无法直接监控，此时需要借助flower

celery -A djangoProject1 flower --prot=5555 --broker_api=http://MQ账号:密码@ip:15672/api/

 新建一个终端执行，成功后http://localhost:5555/查看

最后是beat，首先需要在 settings  INSTALLED_APPS 里注册  'django_celery_beat' ；之后django_celery_beat  创建表 ；最后启动beat   celery -A djangoProject1 beat -l info  

在页面上创建定时任务试试：

任务没有执行，表单中填写的信息已经记录进数据库了。但是我们没有给CELERY_BEAT指明用哪一个数据库的数据。所以还需要在 settings中加入 ：

CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'

之后重启beat，发现自建的任务识别到了

配合rabbitMQ的使用

了解部分使用规则，首先新建一个队列

在队列中手动推送一条消息，然后截取查看

现在可以做一个小实验，先停掉celery worker的进程（停用后创建的消息就不会被自动消费了）。控制台在调用一次task（相当于生产者创建了一个数据，等待进入队列被消费） 之后rabbit页面直接进入celery的列队手动获取消息。可以看到我们task的传参

 此时再启动worker单线程执行下，celery -A djangoProject1 worker -P eventlet -l info --pool=solo 

再来观察，发现刚刚 Ready的变成 Unackde（相当于等待消费者消费后应答的状态）了