Celery定时任务

# 任务队列的链接地址(变量名必须叫这个)
broker_url = ‘redis://127.0.0.1:6379/14‘
# 结果队列的链接地址(变量名必须叫这个)
result_backend = ‘redis://127.0.0.1:6379/15‘

from mycelery.main import app

@app.task(name="test")  # name指定任务名称
def test():
    print(888)

from celery import Celery
from celery.schedules import crontab

# 首先实例化一个celery对象
app = Celery(‘vips5‘)
# 然后配置此对象
app.config_from_object(‘mycelery.config‘)
# # 通过celery对象.autodiscover_tasks 让woker自动去任务队列中发现任务
app.autodiscover_tasks([‘mycelery.auto_send_call‘])  # auto_send_call为存放tasks.py文件的目录

# tesk:test test为任务名称
app.conf.beat_schedule = {
    ‘auto_send‘: {
        ‘task‘: ‘test‘,
        ‘schedule‘: crontab(minute=‘*/1‘),
        ‘args‘: ()
    }
}

celery -A mycelery.main worker -l info -P eventlet

celery -A mycelery.main beat -l info