Django3.2整合celery

创建于 2年前 / 阅读数 0 / 评论数 0 / 2年前

一、各个组件版本：

django 3.2.6
celery 5.1.2
django-celery-beat 2.2.1
django-celery-results 2.2.0

二、在django项目的目录路径<注意不是根目录>下添加celery.py

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

# set the default Django settings module for the 'celery' program.
# 设置django环境变量
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'drf_ops.settings')

app = Celery('drf_ops')

#  使用CELERY_ 作为前缀，在settings中写配置
app.config_from_object('django.conf:settings', namespace='CELERY')

# 从所有已注册的Django应用配置中加载任务模块。从而使celery 能够自动发现这些任务
# 发现任务文件每个app下的task.py
app.autodiscover_tasks()


@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

三、在init.py加入以下内容<路径同上>

from .celery import app as celery_app

# 确保应用程序总是能被找到
# Django启动，shared_task将使用这个应用程序。
__all__ = ['celery_app']

四、在setting.py加入以下配置

这里需要注意的地方，当USE_TZ = True和TIME_ZONE: 'Asia/Shanghai'的时候，下面的CELERY_ENABLE_UTC是看似不起作用的，写入数据库的时间戳还是utc的，并且打开admin后台任务列表的时候如果有数据会提示Database returned an invalid datetime value. Are time zone definitions for your database installed?，这情况要嘛在数据库连接信息加上TIME_ZONE: 'Asia/Shanghai'，要嘛把USE_TZ = True改成False，然后就是全局的时间戳就是按TIME_ZONE = 'Asia/Shanghai' 来写入了，旧数据则需要手动处理

INSTALLED_APPS = [
	...
    'django_celery_results',
    'django_celery_beat',
	...
]
# --------------- Celery 配置 start ------------
CELERY_BROKER_URL = 'redis://:password@192.168.123.13:6379/0' # Broker配置，使用Redis作为消息中间件
# CELERY_RESULT_BACKEND = 'redis://10.1.210.69:6379/0' # 定时任务执行结果存储方式，这里使用redis
CELERY_RESULT_BACKEND = 'django-db' # 定时任务执行结果存储方式
CELERY_RESULT_SERIALIZER = 'json' # 结果序列化方案
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
CELERY_TIMEZONE = 'Asia/Shanghai'  # 使用亚洲/上海时区
CELERY_ENABLE_UTC = False  # 不使用国际标准时间
DJANGO_CELERY_BEAT_TZ_AWARE = False  # 解决时区问题
# 设置结果的保存时间
CELERY_TASK_RESULT_EXPIRES = 10 * 60
# ------------------  end  --------------------

五、编写定时器任务，在每个django APP路径下新建tasks.py，名字固定是这个不能随意改否则无法识别到

from celery import shared_task


@shared_task
def test_celery_execute(x, y):
    print("xxx:", x, "   yyyy:", y)
    return x + y

六、启动启动beat和worker

-A 后面的drf_ops参数为项目名称，和前面定义的一致，按需修改

(venv) [root@localhost drf_ops]# celery -A drf_ops beat -l debug    # 启动beat监听任务变化 debug输入可以看到任务是否报错
(venv) [root@localhost drf_ops]# celery -A drf_ops worker -l debug     # 启动worker执行任务

(venv) [root@localhost drf_ops]# celery -A drf_ops worker  --beat -l debug      # 开发环境可以使用这个命令同时启动beat和worker，生产环境建议不要这样用

七、打开django admin后台添加任务测试即可

讨论数量: 0