Celery任务调度示例

1.Celery任务调度简单示例

环境：Linux，Celery，Flower

脚本：tasks.py

功能：任务调度；任务运行状态监控和日志显示

　tasks.py代码如下：

from celery import Celery,platforms

app = Celery('tasks',backend='amqp',，broker='amqp://public:qfZjSmJjDs51EyIk@192.168.12.103:5672/')
#backend:消息中间件类型，可无；broker：指定AMQP Broker(Advanced Message Queue Protocal，高级消息队列协议 消息中间件)
platforms.C_FORCE_ROOT = True       #用户解决root用户无法启动worker的问题

@app.task                           #对函数做celery task注解
def add(x,y):
return x + y

　　在tasks.py同级目录执行celery flower -A tasks –port=8083，启动worker，绑定flower监控并设置监控端口；或者开两个进程分别执行celery -A tasks worker -l info和celery -A tasks flower –port=8083，效果也一样。

　　worker启动后，另开进程进入task.py目录，执行如下代码，会在localhost:8083的监控页面看到tasks运行结果。　

>>> from tasks import add
>>> add.delay(2,34)
<AsyncResult: 61df6787-aa68-4c0a-8a9d-3a9ce52be4df>
>>>

参考文献：http://flower.readthedocs.org/en/latest/install.html#usage

二.Celery定时任务示例

脚本：tasks.py，config.py（配置文件）

功能：Crontab定时，Periodic Task

tasks.py代码如下：

from celery import Celery,platforms

app = Celery('tasks')
app.config_from_object('config')    #以config.py作为配置文件导入参数
platforms.C_FORCE_ROOT = True

@app.task
def add(x,y):
return x + y

　　config.py代码如下：

from __future__ import absolute_import          #如果没有这一行，下一行可能会出错
from celery.schedules import crontab
from datetime import timedelta

BROKER_URL = 'amqp://public:qfZjSmJjDs51EyIk@192.168.12.103:5672/'

# Crontab定时任务的设置方式
# CELERYBEAT_SCHEDULE = {
#     'every-minute': {
#         'task': 'tasks.add',
#         'schedule': crontab(minute='*/1'),       #crontab的参数设置见后面
#         'args': (1,2),
#     },
# }

# Periodic task的设置方式
CELERYBEAT_SCHEDULE = {
'add-every-2-seconds': {
'task': 'tasks.add',
'schedule': timedelta(seconds=2),
'args': (16, 10),
},
}

CELERY_TIMEZONE = 'UTC'  #时区设置，也可以为'Europe/London'

　　在脚本同级目录执行celery -A tasks worker -B，即启动worker和beat服务；或者先用celery -A proj worker –loglevel=INFO启动worker，再用celery -A tasks beat -s celerybeat-schedule #这里的celerybeat-schedule指定一个记录文件**启动beat服务也行。

　　定时任务的运行代码如下：　

celery beat v3.1.20 (Cipater) is starting.
__    -    ... __   -        _
Configuration ->
. broker -> amqp://public:**@192.168.12.103:5672//
. loader -> celery.loaders.app.AppLoader
. scheduler -> celery.beat.PersistentScheduler
. db -> celerybeat-schedule
. logfile -> [stderr]@%INFO
. maxinterval -> now (0s)
[2016-02-29 15:23:48,893: INFO/MainProcess] beat: Starting...
[2016-02-29 15:24:00,001: INFO/MainProcess] Scheduler: Sending due task every-minute (tasks.add)
[2016-02-29 15:25:00,052: INFO/MainProcess] Scheduler: Sending due task every-minute (tasks.add)
[2016-02-29 15:26:00,051: INFO/MainProcess] Scheduler: Sending due task every-minute (tasks.add)
[2016-02-29 15:27:00,037: INFO/MainProcess] Scheduler: Sending due task every-minute (tasks.add)

　　在localhost:8083查看tasks运行状态与详细日志，如下：

　　


　　


　　Crontab的参数设置如下，详细见Crontab参数设置

　　


　　此外，如果上一个task执行时间大于任务的时间间隔，则会造成任务重叠（Overlap），如果要避免此现象，除了延长任务的时间间隔，还可以通过锁实现http://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html#cookbook-task-serial

参考文献：

https://gist.github.com/alexex/3018527

http://docs.celeryproject.org/en/latest/userguide/periodic-tasks.html

http://www.metaltoad.com/blog/celery-periodic-tasks-installation-infinity

http://my.oschina.net/hochikong/blog/419191?p={{currentPage-1}}