Django + celery + django-celery-result + django-celery-beat 实现定时任务-摩杜云开发者社区

1.库版本
celery 5.3.4
Django 4.2.6
django-celery-beat 2.5.0
django-celery-results 2.5.1
django-timezone-field 6.0.1
eventlet 0.33.3
flower 2.0.1
redis 3.5.3
项目名称： new_project

2.settings.py 配置(时区有关系)

LANGUAGE_CODE = 'zh-hans'
TIME_ZONE = 'Asia/Shanghai'
USE_I18N = True

# django-celery  配置的部分
# Broker配置，使用Redis作为消息中间件
CELERY_BROKER_URL = 'redis://127.0.0.1:6379/0'

# BACKEND配置，可以选用redis或存放数据库
# CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
CELERY_RESULT_BACKEND = 'django-db'  # 存放至数据库

CELERY_CACHE_BACKEND = 'default'  # 设置缓存为默认缓存,
CELERY_TASK_TRACK_STARTED = True  # 设置任务启动追踪
CELERY_TASK_TIME_LIMIT = 30 * 60  # 设置任务运行时间限制为30分钟

# 结果序列化方案
CELERY_RESULT_SERIALIZER = 'json'

# 任务结果过期时间，秒
CELERY_TASK_RESULT_EXPIRES = 60 * 60 * 24

# 时区配置
CELERY_TIMEZONE = 'Asia/Shanghai'

# # 指定导入的任务模块，可以指定多个
CELERY_IMPORTS = (
   'new_project.tasks',
)

# django-celery-beat配置
# SCHEDULER 定时任务保存数据库
# 将任务调度器设为DatabaseScheduler
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
# 为存储结果设置过期日期，默认1天过期。如果beat开启，Celery每天会自动清除。
# 设为0，存储结果永不过期
# CELERY_RESULT_EXPIRES = xx
CELERY_TASK_RESULT_EXPIRES = 60*60*24  # 后端存储的任务超过一天时，自动删除数据库中的任务数据，单位秒
CELERY_MAX_TASKS_PER_CHILD = 1000  # 每个worker执行1000次任务后，自动重启worker，防止任务占用太多内存导致内存泄漏
DJANGO_CELERY_BEAT_TZ_AWARE = False

　　3.celery.py(与settings.py同目录)

import os
from datetime import timedelta

import django
from celery import Celery

# 设置系统环境变量，安装django，必须设置，否则在启动celery时会报错
# djangoProject1.settings
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'new_project.settings')
django.setup()

celery_app = Celery('new_project')
celery_app.config_from_object('django.conf:settings', namespace='CELERY')
# 发现每个app下的task.py
celery_app.autodiscover_tasks()

# 手动设置定时执行任务(也可参考后面通过admin后台设置任务)
CELERY_BEAT_SCHEDULE = {
    "add-every-30s": {
        "task": "new_project.tasks.show_msg",
        'schedule': timedelta(seconds=20),
        # 'args': (3,)  # 传递参数-
    },
}

　　4.配置定时任务task.py(与settings.py文件同目录)

import random
from pdf2docx import Converter
from celery import shared_task

@shared_task
def show_msg():
    print("================================")
    return 111
该任务可在admin/页面定时任务中被发现，且执行结果为return的值111，在worker日志中可以看到print的值和return的值

5.迁移数据库，会生成celery-result和celery-beat相关的表
6.启动Django服务:python manage.py runserver 8000
7.启动celery-worker:celery -A new_project worker -l debug -P eventlet
8.启动celery-beat: celery -A new_project beat -l info

9.定时任务：
如果第3点手动配置了任务，则任务会自动运行，若未配置可以手动配置
登录后台admin/：使用admin账号进行登录
登录后点击"周期性任务"后面的增加进行定时任务增加，设置相关信息，其中任务已注册的选项来源于所有app里面task.py里面的方法

Django + celery + django-celery-result + django-celery-beat 实现定时任务

10.建议开始时间选择从00:00:00开始，保存后定时任务开始执行，可以在worker日志中查看任务执行情况

遇到的问题：
1.Django从3切到4版本，无法识别django.urls库导入url
解决：把from django.conf.urls import url 修改为 from django.urls import re_path as url

2.定时任务可在admin/页面手动执行，但是定时任务在worker中未执行
解决：应该是时间问题，把任务时间调整到从00:00:00开始后，任务正常执行