定時任務

Celery 中啟動定時任務有兩種方式，（1）在配置檔案中指定；（2）在程式中指定。

# cele.py
import celery


app = celery.Celery('cele', broker='redis://localhost:6379')


@app.task
def send(message):
    return message


app.conf.beat_schedule = {
    'send-every-10-seconds': {
        'task': 'cele.send',
        'schedule': 10.0,
        'args'
 
: ('Hello World', )
    },
}

可以通過在配置檔案中編寫 beat_schedule 屬性，來配置週期性任務，上面的示例配置了一個每十秒執行一次的週期任務，任務為 cele.send，引數為 ‘Hello World’。當然你也可以將這個配置寫到單獨的配置檔案中進行讀取。這種配置的方式可以支援多個引數，

• task： 指定任務的名字
• schedule : 設定任務的排程方式，可以是一個表示秒的整數，也可以是一個 timedelta 物件，或者是一個 crontab 物件（後面介紹），總之就是設定任務如何重複執行
• args： 任務的引數列表
• kwargs：任務的引數字典
• options：所有 apply_async 所支援的引數

同時官方文件中也指出，可以通過下面這種方式對定時任務進行設定。

from celery import Celery
from celery.schedules import crontab

app = Celery()

@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 提供了一個 crontab 的物件，可以對於定時任務進行更為精確的時間設定，而不僅限於多少秒重複一次這種簡單的任務。下面用例子的方式來說明一下。更詳細的說明可以看這裡 crontab scheduler

from celery.schedules import crontab

# 每分鐘執行一次
c1 = crontab()

# 每天凌晨十二點執行
c2 = crontab(minute=0, hour=0)

# 每十五分鐘執行一次
crontab(minute='*/15')

# 每週日的每一分鐘執行一次
crontab(minute='*',hour='*', day_of_week='sun')

# 每週三，五的三點，七點和二十二點沒十分鐘執行一次
crontab(minute='*/10',hour='3,17,22', day_of_week='thu,fri')

到目前為止，只是對任務進行了配置，但是還沒有實際執行任務，要支援週期任務，需要啟動一個元件 beat，它用於對任務進行排程，我們以 cele.py 為例進行說明。

celery -A cele beat

這個命令會啟動 cele 應用的 beat，當然也可以在啟動 worker 的時候使用 -B 引數同時啟動 beat 。

celery -A cele worker -l info -B

這樣就可以在啟動的視窗看到每十秒執行一次的 send 任務了。