Django 2.1.7 Celery 4.3.0 Periodic Tasks 周期性任务调度

原文链接：Django 2.1.7 Celery 4.3.0 Periodic Tasks 周期性任务调度

任务调度器 beat

celery beat是一个调度器，它可以周期内指定某个worker来执行某个任务。如果我们想周期执行某个任务需要增加beat_schedule配置信息.

在celeryconfig.py增加调度内容

# 设置结果存储
CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/9'# 设置代理人broker
BROKER_URL = 'redis://127.0.0.1:6379/8'# 配置任务路由
CELERY_ROUTES=({'celery_tasks.tasks.my_task5': {'queue': 'queue1'},'celery_tasks.tasks.my_task6': {'queue': 'queue1'},'celery_tasks.tasks.my_task7': {'queue': 'queue2'},},
)# 配置周期性任务，　或者定时任务
CELERYBEAT_SCHEDULE = {'every-5-seconds':{'task': 'celery_tasks.tasks.my_task1','schedule': 5.0,'args': (10, 20),}
}

tasks.py模块内容如下:

from celery_tasks.celery import app as celery_app# 设置周期性任务
@celery_app.task
def my_task1(a, b):print("my_task1任务正在执行....")return a + b@celery_app.task
def my_task2(a, b):print("my_task2任务正在执行....")return a + b@celery_app.task
def my_task3(a, b):print("my_task3任务正在执行....")return a + b@celery_app.task
def my_task4(a, b):print("my_task3任务正在执行....")return a + b@celery_app.task
def my_task5():print("my_task5任务正在执行....")@celery_app.task
def my_task6():print("my_task6任务正在执行....")@celery_app.task
def my_task7():print("my_task7任务正在执行....")

启动woker处理周期性任务:

启动周期性任务是无法在windows上运行的，我在Centos7上启动，只要能够访问作为broker的redis，那么就可以执行任务。

celery -A celery_tasks worker -l info --beat

启动日志如下：

当启动了之后，每隔5秒就执行一次my_task1,如下：

配置cron任务

如果我们想指定在某天某时某分某秒执行某个任务，可以执行cron任务, 增加配置信息如下:

from celery.schedules import crontab# 配置周期性任务，　或者定时任务
CELERYBEAT_SCHEDULE = {'every-5-seconds':{'task': 'celery_tasks.tasks.my_task1','schedule': 5.0,'args': (10, 20),},'add-every-monday-morning': {'task': 'celery_tasks.tasks.my_task1','schedule': crontab(hour=7, minute=30, day_of_week=1),'args': (16, 16),},
}

crontab例子: http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

开启一个celery beat服务：
celery -A celery_tasks worker -l info --beat

celery需要保存上次任务运行的时间在数据文件中，文件在当前目录下名字叫celerybeat-schedule.
beat需要访问此文件则执行以下命令：

celery -A celery_tasks beat -s celerybeat-schedule