欢迎访问宙启技术站
智能推送

使用celery.schedulescrontab()制定每周定时任务的计划

发布时间:2023-12-23 23:02:51

Celery是一个分布式任务队列,可以实现任务调度和分发。通过使用Celery的schedules模块中的crontab函数,可以制定每周定时任务的计划。

首先,需要安装Celery和cryptography依赖库:

pip install celery cryptography

然后,导入相关的模块和函数:

from celery import Celery
from celery.schedules import crontab

接下来,创建一个Celery对象:

app = Celery('example', broker='pyamqp://guest@localhost//')

然后,通过使用crontab函数制定每周定时任务的计划。crontab函数接受6个参数,分别表示分钟、小时、天、月、星期、年:

schedule = crontab(day_of_week='*/1')

上述代码表示每周的每天执行任务。

然后,定义一个任务函数:

@app.task
def example_task():
    print('This is an example task.')

最后,在Celery对象上设置定时任务:

app.conf.beat_schedule = {
    'example_task': {
        'task': 'example_task',
        'schedule': schedule,
    },
}

完整的示例代码:

from celery import Celery
from celery.schedules import crontab

app = Celery('example', broker='pyamqp://guest@localhost//')

schedule = crontab(day_of_week='*/1')

@app.task
def example_task():
    print('This is an example task.')

app.conf.beat_schedule = {
    'example_task': {
        'task': 'example_task',
        'schedule': schedule,
    },
}

上述代码中,创建了一个Celery对象,并设置了一个每周每天的定时任务。任务函数example_task打印一条信息。定时任务的设置通过beat_schedule属性完成,其中example_task是任务的名称,task表示要执行的任务函数,schedule表示任务的执行计划。

在运行Celery的worker和beat进程之后,每周的每天都会执行example_task任务。

总结来说,使用Celery的schedules模块中的crontab函数可以制定每周定时任务的计划。通过在Celery对象上设置beat_schedule属性,可以将定时任务与任务函数关联起来。