示例代码
创建schedule_tasks.py文件
from celery import Celery
from celery.schedules import crontab
app = Celery("SchedulerTasks", broker=‘pyamqp://admin:[email protected]:5672//‘)
@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
# Calls test(‘hello‘) every 10 seconds.
sender.add_periodic_task(10.0, test.s(‘hello‘), name=‘add every 10‘)
# Calls test(‘world‘) every 30 seconds
sender.add_periodic_task(30.0, test.s(‘world‘), expires=10)
# Executes every Monday morning at 7:30 a.m.
sender.add_periodic_task(
crontab(hour=7, minute=30, day_of_week=1),
test.s(‘Happy Mondays!‘),
)
@app.task
def test(arg):
print(arg)
启动Celery Beat-任务生产者
# celery -A schedule_tasks beat
启动Celery Worker-任务消费者
# celery -A schedule_tasks worker
原文地址：https://www.cnblogs.com/vincenshen/p/12315132.html
时间： 2024-10-09 15:17:51