Celery定時任務

celery支持定時任務，設定好任務的執行時間，celery就會定時自動執行， 這個定時任務模塊叫celery beat

 

示例

#cat test.py

from celery import Celery
from celery.schedules import crontab

app = Celery(
            broker='redis://10.10.10.11',
            backend='redis://10.10.10.11'
)

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):

    sender.add_periodic_task(10.0, test_celery.s('hello'), name='add every 10')    #每10秒　　test_celery爲下面定義的函數，‘hello’爲傳入的參數

    sender.add_periodic_task(30.0, test_celery.s('world'), expires=10)               #每30秒發送一次

    sender.add_periodic_task(

        crontab(hour=13, minute=50, day_of_week=3),      #每週一7點半

        test_celery.s('Happy Mondays!'),
    )

@app.task
def test_celery(arg):
    print(arg)

 

啓動監聽和worker

D:\django-project\wechat>celery -A test  beat　　　　　　　　　　　　#啓動任務調度器 celery beat，test爲要執行的文件名
D:\django-project\wechat>celery -A test  worker -P eventlet　　　　#啓動worker，執行調度器發起的任務

 

使用crontab作定時任務

from celery.schedules import crontab

app.conf.beat_schedule = {

    'add-every-monday-morning': {
        'task': 'tasks.add',　　　　　　　　　　　　　　　　　　　　　　　　#task.add （爲task.py下的add函數） 'schedule': crontab(hour=7, minute=30, day_of_week=1),
        'args': (16, 16),　　　　　　　　　　　　　　　　　　　　　　　　　　#args 爲傳入的參數
    },
}

 

示例

from celery import Celery
import subprocess

from celery.schedules import crontab

app = Celery("test",
            broker='redis://10.10.10.11',
            backend='redis://10.10.10.11'
)

app.conf.beat_schedule = {

    # Executes every Monday morning at 7:30 a.m.

    'add-every-monday-morning': {

        'task': 'test.cmd_run',

        'schedule': ,

        'args': ('ipconfig',),

    },

}
app.conf.timezone = 'UTC'

@app.task
def cmd_run(cmd):
    result = subprocess.Popen(cmd,shell=True,stdout=subprocess.PIPE,stderr=subprocess.PIPE)
    return result.stdout.read().decode("utf-8")crontab(hour=7, minute=30, day_of_week=1)

 

重啓celery和監聽