django-celery-beat自动调度异步任务

Celery是一个简单、灵活且可靠的分布式系统，专门用于处理大量消息的实时任务调度。它支持使用任务队列的方式在分布的机器、进程、线程上执行任务调度。Celery不仅支持异步任务（如发送邮件、文件上传、图像处理等耗时操作），还支持定时任务，即需要在特定时间执行的任务。Celery本身不提供消息服务，需要借助RabbitMQ、Redis等消息中间件，本案例使用的是Redis。

Celery Beat则是Celery的一个组件，专门用于处理定时任务调度。它包含一个调度器，负责根据配置的时间表计划任务的执行。这些任务通常是Celery任务，即异步执行的函数或方法。Celery Beat将计划的任务发送到Celery任务队列，由Celery Worker处理并执行队列中的任务。此外，Celery Beat还支持任务的持久性，即使在系统重启后也能够保持已计划的周期性任务。

开发环境：Python3 + MySQL + Redis + PyCharm专业版

一、创建Django项目

参考 Python框架Django入门教程-CSDN博客前三步

二、安装celery、mysql、redis等依赖包

eventlet 是一个python协程模板，celery 4版本以上在windows环境进行测试需要安装此依赖

django_celery_results 是任务执行结果的依赖

mysqlclient
redis
celery
eventlet
django-celery-beat
django_celery_results

三、初始化Celery数据库

打开PyCharm的终端，执行以下命令

python manage.py makemigrations
python manage.py migrate

打开数据库查看，执行命令后自动创建了一些表

django_celery_beat_clockedschedule  # 以指定时间执行任务，例如：2024-05-22 09:22:10
django_celery_beat_crontabschedule  # 以crontab格式时间执行任务，某月某天星期几某时某分
django_celery_beat_intervalschedule  # 以间隔时间执行任务，例如：每5秒、每2小时
django_celery_beat_periodictask  # 存储要执行的任务。
django_celery_beat_periodictasks  # 索引和跟踪任务更改状态
django_celery_beat_solarschedule  # 以天文时间执行任务，例如：日出、日落
django_celery_results_chordcounter  # 存储Celery的chord任务的状态
django_celery_results_groupresult  # 存储Celery的group任务的结果
django_celery_results_taskresult  # 存储Celery任务的执行结果

四、配置Celery

修改（注意是修改，不是添加！！！）settings.py，找到 INSTALLED_APPS变量（约31行），将celery注册到django的应用管理中

在settings.py末尾添加celery配置：

CELERY_BROKER_URL = 'redis://localhost:6379/0'  # 使用Redis作为消息代理
CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'  # 结果存储也使用Redis
# 配置 celery 定时任务使用的调度器，使用django_celery_beat插件用来动态配置任务
CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
# 配置celery自动存储任务执行结果
CELERY_RESULT_BACKEND = 'django_celery_results.backends:DatabaseBackend'
CELERY_TIMEZONE = 'Asia/Shanghai'  # 设置时区
# 是否启用UTC
CELERY_ENABLE_UTC = False
# 是否开启时间感知
DJANGO_CELERY_BEAT_TZ_AWARE = False

在settings.py同级目录下创建celery.py，然后添加以下内容：

import os

from celery import Celery
from django.conf import settings

# djangoDemo是项目名，大家根据自己的情况进行替换！！！
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'djangoDemo.settings')
# djangoDemo是项目名，大家根据自己的情况进行替换！！！
app = Celery('djangoDemo')
# 从django的设置中读取配置信息
app.config_from_object('django.conf:settings', namespace='CELERY')
# 自动发现app下的任务
app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)

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

五、创建应用模块，进行测试

打开PyCharm终端执行命令，创建应用模块，这里命名为celeryapp，将新的应用模块注册到django的应用管理中

python manage.py startapp celeryapp

在新的应用模块中创建tasks.py，添加异步函数，文件名必须是tasks.py，否则后面启动Celery的时候监听不到

@shared_task
def task_one():
    print("------------------------- 000 <<<")
    # 业务逻辑...
    print("------------------------- 111 <<<")
    return "222"

在新的应用模块中修改views.py，添加一个测试接口。这里说一个坑：在settings.py中配置了TIME_ZONE = 'Asia/Shanghai' 和 USE_TZ = True 之后，通过datetime.now()获取的是亚洲上海时间，但是把这个时间存到数据库，就会自动减少8小时，变成了UTC时间，就很无语，于是我把USE_TZ的值改为False，发现存到数据库中的时间变成正常的亚洲上海时间。然而Celery定时执行任务的时区是UTC，经过多次测试，配置了CELERY_TIMEZONE等时区相关的配置，发现好像并没什么用，Celery定时执行任务的时区依然是UTC，无奈只能把任务执行的时间减少8小时

import json
from datetime import datetime, timedelta

from django.http import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django_celery_beat.models import ClockedSchedule, PeriodicTask

def celery_test(request):
    # 任务名
    task_name = 'celery_test'
    # 任务执行的时间，设置为下一分钟的10秒，celery执行时间的时区是UTC，要保证数据库中的时间是UTC时间，这里获取的是亚洲上海的时间，所以减了8小时
    time = (datetime.now() - timedelta(hours=8) + timedelta(minutes=1)).strftime("%Y-%m-%d %H:%M:10")
    # 创建任务的执行时间
    clock = ClockedSchedule.objects.get_or_create(clocked_time=time)
    # 创建指定时间执行的celery任务
    PeriodicTask.objects.update_or_create(
        name=task_name,  # 任务名，尽量保证唯一性，若该任务已存在，则更新该任务
        task='celeryapp.tasks.task_one',  # 要执行的异步函数的全路径
        defaults={
            'clocked': clock[0],  # 使用clocked在指定时间执行该任务
            'one_off': True,  # 在任务执行完一次后关闭该任务
            'enabled': True,  # 开启任务
            'args': json.dumps([]),  # 参数列表，必须是json格式的数组
        }
    )
    return JsonResponse({'message': '200'})

修改urls.py，在urlpatterns中添加路由

from celeryapp import views

urlpatterns = [
    # ......
    path('celery/test/', views.celery_test),
]

六、启动项目进行测试

先启动Django项目，然后在分别两个终端中执行命令启动Celery和Celery Beat，命令中的djangoDemo是项目名，大家根据自己的情况进行替换

celery -A djangoDemo worker -P eventlet -l info  # 启动worker监听异步任务
celery -A djangoDemo beat -l info  # 启动beat任务调度器

这里的woker已经监听到我们创建的celeryapp.tasks.task_one异步函数了

这里说明beat启动成功了：

浏览器输入请求地址：http://127.0.0.1:8000/celery/test/ ，创建Celery任务

接口请求成功后查看数据库，django_celery_beat_periodictask表新增了一条异步任务，其中的clocked_id字段指向django_celery_beat_clockedschedule表，该表新增了一条任务的执行时间。celery.backend_cleanup是Celery自动创建的，不用管

在到达任务执行时间后观察woker和beat的终端日志

查看beat终端日志，红框第一行是我们发送请求成功创建异步任务之后，CeleryBeat已经检测到数据库中有任务发生变化（CeleryBeat每5秒检测一次，使用debug级日志可查看到）；第二行CeleryBeat将一个名为celery_test的任务发送给worker，让woker执行celeryapp.tasks.task_one异步函数，消费该任务

在woker终端的日志中可以看到任务执行的结果：

七、 Celery Beat常用的三种时间控制器

clockedSchedule：指定某个时间执行任务，例如：2024-05-22 09:22:10，对应的表是django_celery_beat_clockedschedule，该表仅有id和clocked_time两个字段

crontabSchedule：指定crontab格式的时间执行任务，某月某天星期几某时某分，与linux的定时任务规则一致，可参考Linux定时任务-CSDN博客，对应的表是django_celery_beat_crontabschedule，该表有7个字段

  `id` 
  `minute` 分钟
  `hour` 小时
  `day_of_week`  星期几
  `day_of_month`  每月的哪些天
  `month_of_year`  每年的哪些月份
  `timezone`  时区

intervalSchedule：间隔指定时间执行任务，对应的表是django_celery_beat_intervalschedule，该表有3个字段id、every（间隔时长）、period（时间单位，可选时、分、秒、微秒、天）

代码示例，clockedSchedule上面已经演示过了，这里只演示另外两种：

# crontabSchedule
def celery_test2(request):
    task_name = 'celery_test2'
    crontab = CrontabSchedule.objects.get_or_create(
        minute='*/1',  # 每1分钟
        hour='*',  # 每小时
        day_of_week='*',  # 一周中的哪几天，*表示每天
        day_of_month='*',  # 月份中的哪一天，*表示每一天
        month_of_year='*',  # 年中的哪一月，*表示每个月
        timezone='Asia/Shanghai'
    )

    PeriodicTask.objects.update_or_create(
        name=task_name,
        task='celeryapp.tasks.task_one',  # Celery任务的全路径
        defaults={
            'crontab': crontab[0],  # 使用crontab格式时间执行该任务
            'one_off': False,  # 在任务执行完一次后关闭该任务
            'enabled': True,  # 开启任务
            'args': json.dumps([]),
        }
    )

    return JsonResponse({'message': '200'})


# intervalSchedule
def celery_test3(request):
    task_name = 'celery_test3'

    interval = IntervalSchedule.objects.get_or_create(
        every=1,  # 间隔时间
        period=IntervalSchedule.MINUTES,  # 周期单位，这里是分钟
    )

    PeriodicTask.objects.update_or_create(
        name=task_name,
        task='celeryapp.tasks.task_one',  # Celery任务的全路径
        defaults={
            'interval': interval[0],  # 使用interval间隔指定时间执行该任务
            'one_off': False,  # 在任务执行完一次后关闭该任务
            'enabled': True,  # 开启任务
            'args': json.dumps([]),
        }
    )

    return JsonResponse({'message': '200'})

其实只要创建不同的时间控制器，然后在创建任务的时候作为参数放进去即可，注意一个任务只能使用一种时间控制器

参考文献：

https://blog.csdn.net/wuwei_201/article/details/129650089

https://blog.51cto.com/u_15703497/6252757