celery週期性任務

celery提供了一個叫celery beat的服務，用於定時驅使worker執行任務。也就是說，若是本地沒有活動的worker，它將不會獲得任何執行結果，他只是負責把任務消息傳到rabbitmq，一旦啓動一個可用worker，則自動從rabbitmq獲取任務信息並執行。

與此配置相關的參數是CELERYBEAT_SCHEDULE，我把個人celery應用proj的全部配置內容都放置在一個config.py文件中：

from __future__ import absolute_import
from datetime import timedelta
from celery.schedules import crontab

CELERY_TASK_RESULT_EXPIRES=3600
CELERY_TASK_SERIALIZER='json'
CELERY_ACCEPT_CONTENT=['json']
CELERY_RESULT_SERIALIZER='json'

CELERYBEAT_SCHEDULE = {
    'add-every-1-min': {
        'task': 'proj.agent.add',
        'schedule': crontab(),
        'args': (16, 16),
    },
}
#CELERYBEAT_SCHEDULE = {
#    'add-every-2-seconds': {
#        'task': 'proj.agent.add',
#        'schedule': timedelta(seconds=3),
#        'args': (16, 16)
#    },
#}

CELERY_TIMEZONE = 'UTC'

目前的定時任務是：

add-every-4-s

task指定了相應的任務：proj目錄下agent模塊的add函數

schedule指定了定時工具，這裏是celery.schedules的crontab

args是任務的參數

 

此時咱們回到proj所在的目錄中，啓動一個worker：

root@workgroup0:~/celeryapp/configtest# ls
celerybeat-schedule  logging  proj

root@workgroup0:~/celeryapp/configtest# celery -A proj worker --loglevel=INFO
/usr/local/lib/python2.7/dist-packages/celery/platforms.py:766: RuntimeWarning: You are running the worker with superuser privileges, which is
absolutely not recommended!

Please specify a different user using the -u option.

User information: uid=0 euid=0 gid=0 egid=0

  uid=uid, euid=euid, gid=gid, egid=egid,
 
 -------------- celery@workgroup0.hzg.com v3.1.17 (Cipater)
---- **** ----- 
--- * ***  * -- Linux-3.13.0-24-generic-x86_64-with-Ubuntu-14.04-trusty
-- * - **** --- 
- ** ---------- [config]
- ** ---------- .> app:         proj:0x7f0027635510
- ** ---------- .> transport:   amqp://guest:**@localhost:5672//
- ** ---------- .> results:     amqp://guest@loaclhost//
- *** --- * --- .> concurrency: 1 (prefork)
-- ******* ---- 
--- ***** ----- [queues]
 -------------- .> celery           exchange=celery(direct) key=celery
                

[tasks]
  . proj.agent.add
  . proj.agent.mul
  . proj.agent.writefile
  . proj.agent.xsum

[2015-05-24 15:00:37,873: INFO/MainProcess] Connected to amqp://guest:**@127.0.0.1:5672//
[2015-05-24 15:00:37,940: INFO/MainProcess] mingle: searching for neighbors
[2015-05-24 15:00:38,980: INFO/MainProcess] mingle: all alone
[2015-05-24 15:00:39,021: WARNING/MainProcess] celery@workgroup0.hzg.com ready.

worker啓動成功，此時再開一個終端，啓動beat服務：

root@workgroup0:~/celeryapp/configtest# celery -A proj beat -s celerybeat-schedule #這裏的celerybeat-schedule指定一個記錄文件

celery beat v3.1.17 (Cipater) is starting.
__    -    ... __   -        _
Configuration ->
    . broker -> amqp://guest:**@localhost:5672//
    . loader -> celery.loaders.app.AppLoader
    . scheduler -> celery.beat.PersistentScheduler
    . db -> celerybeat-schedule
    . logfile -> [stderr]@%INFO
    . maxinterval -> now (0s)
[2015-05-24 15:02:53,761: INFO/MainProcess] beat: Starting...
[2015-05-24 15:03:00,000: INFO/MainProcess] Scheduler: Sending due task add-every-1-min (proj.agent.add)

#已經相隔1min了

[2015-05-24 15:04:00,066: INFO/MainProcess] Scheduler: Sending due task add-every-1-min (proj.agent.add)

返回看看worker的輸出：

[2015-05-24 15:01:50,827: INFO/MainProcess] Task proj.agent.add[9b6f962a-9b66-4fde-916f-fc5a951ad599] succeeded in 0.0342152439989s: {'value': '32'}
[2015-05-24 15:02:24,923: INFO/MainProcess] Received task: proj.agent.add[e4b9840b-09f6-4db6-88c1-2a418b11d393]
[2015-05-24 15:02:24,947: INFO/MainProcess] Task proj.agent.add[e4b9840b-09f6-4db6-88c1-2a418b11d393] succeeded in 0.0200459280004s: {'value': '32'}
[2015-05-24 15:03:00,015: INFO/MainProcess] Received task: proj.agent.add[98f44dd1-e6e2-4457-bfd6-ff59d0ee6d2f]
[2015-05-24 15:03:00,031: INFO/MainProcess] Task proj.agent.add[98f44dd1-e6e2-4457-bfd6-ff59d0ee6d2f] succeeded in 0.0125673500006s: {'value': '32'}

這就是週期性任務的執行。

 

遇到的坑：

在配置文件中，from __future__ import absolute_import這一行很關鍵，若是沒有這一行，

from celery.schedules import crontab

這個命令執行時會報錯，celery beat沒法正常啓動。

 

補充：

默認狀況下，celery beat使用UTC時區，你也能夠配置其餘時區：

CELERY_TIMEZONE = 'Europe/London'

 

關於設置任務執行週期，你能夠經過datetime的timedelta設置，能夠讓任務執行間隔精確到秒，相應的配置以下：

CELERYBEAT_SCHEDULE = {
    'add-every-2-seconds': {
        'task': 'proj.agent.add',
        'schedule': timedelta(seconds=3),
        'args': (16, 16)
    },
}

也能夠用crontab風格的：

CELERYBEAT_SCHEDULE = {
    # Executes every Monday morning at 7:30 A.M
    'add-every-1-min': {
        'task': 'proj.agent.add',
        'schedule': crontab(),
        'args': (16, 16),
    },
}

 

關於一個CELERYBEAT_SCHEDULE的能夠配置的參數，以及crontab的詳細示例，請參見celery官方文檔。

 

關於啓動celery beat的tips，我這裏只貼原文：

Starting the Scheduler

To start the celery beat service:

$ celery -A proj beat

You can also start embed beat inside the worker by enabling workers -B option, this is convenient if you will never run more than one worker node, but it’s not commonly used and for that reason is not recommended for production use:

$ celery -A proj worker -B

Beat needs to store the last run times of the tasks in a local database file (namedcelerybeat-schedule by default), so it needs access to write in the current directory, or alternatively you can specify a custom location for this file:

$ celery -A proj beat -s /home/celery/var/run/celerybeat-schedule