在项目中使用celery
项目结构：
proj/__init__.py # 注意这个文件 django是不带的默认 对应django的项目
/celery.py
/tasks.py
proj/celery.py
from __future__ import absolute_import, unicode_literals
from celery import Celery
app = Celery('proj',
broker='amqp://', # 使用的消息队列
backend='amqp://', # 使用的结果存储
include=['proj.tasks']) # 任务们
if __name__ == '__main__':
app.start()
proj/tasks.py
from __future__ import absolute_import, unicode_literals
from .celery import app
@app.task
def add(x, y):
return x + y
启动工作进程：
celery -A proj worker -l info
-------------- celery@halcyon.local v4.0 (latentcall)
---- **** -----
--- * *** * -- [Configuration]
-- * - **** --- . broker: amqp://guest@localhost:5672//
- ** ---------- . app: main:0x1012d8590
- ** ---------- . concurrency: 8 (processes)
- ** ---------- . events: OFF (enable -E to monitor this worker)
- ** ----------
- *** --- * --- [Queues]
-- ******* ---- . celery: exchange:celery(direct) binding:celery
--- ***** -----
[2012-06-08 16:23:51,078: WARNING/MainProcess] celery@halcyon.local has started.
broker就是配置中的消息队列，concurrency就是工作进程数量（默认是cpu数量）如果所有进程都被占用了那么新的任务需要等待，events是指定celery是否发送监控信息，queue就是队列。
终止工作进程
直接control -c（按照上述命令在前台启动的），当然了如果在后台启动和停止：
celery multi start w1 -A proj -l info
celery multi restart w1 -A proj -l info
# 异步关闭 立即返回
celery multi stop w1 -A proj -l info
# 等待关闭操作完成
celery multi stopwait w1 -A proj -l info
默认会在当前目录下创建pid和log文件，指定路径和名称：
celery multi start w1 -A proj -l info --pidfile=/var/run/celery/%n.pid \
--logfile=/var/log/celery/%n%I.log
关于-A（--app）选项：
指定celery app，可以用 module.path:attribute的形式指定，也可直接指定package，如--app=proj，然后会搜索其中的proj.app，proj.celery等。
操作已被创建的worker时，不需要参数什么的完全一样，只要pidfile和logfile一样即可。
调用
add.delay(1, 1)
add.apply_async((1, 1))
add.apply_async((2, 2), queue='lopri', countdown=10)
# 指定要发送到哪个队列 运行时间延迟countdown
# 结果获得
res = add.delay(1, 1)
res.get()
celery默认不产生结果的原因：因为各种应用的需求不同，而大多数任务保存返回值没有什么意义，而且产生结果并不是用来监控任务和工作进程，应该是使用event消息的专门的监控模块。
res.id # 任务id uuid
# 任务出错会raise 错误 可指定propagate
res.get(propagate=False)
res.failed()
res.successful()
res.state
# PENDING -> STARTED -> SUCCESS(FAILURE)
# STARTED 需要设置 task_track_started 或在任务级别设置@task(track_started=True)
# 重试的情况也有
# PENDING -> STARTED -> RETRY -> STARTED -> RETRY -> STARTED -> SUCCESS
工作流
有时想把一个任务的签名（包含了实参和任务的执行选项）发送到另一个进程或把参数发送到另一个函数。
add.signature((2, 2), countdown=10) # with args and kwargs
add.s(2, 2) # with args only
然后获得的签名就可以调用：
s1 = add.s(2, 2)
res = s1.delay()
res.get()
# 也可以部分指定参数 然后在调用时补全参数
s2 = add.s(2)
res = s2.delay(8)
res.get()
调用方式
之前一直使用的是delay(*args, **kwargs)的方式来调用，还有apply_async的调用方式，支持一些执行时的选项
Groups
group是同时调用的一系列任务，返回特殊的结果可以获得组内所有的执行结果，可以按照任务的顺序从该结果中获得对应任务的结果。
group(add.s(i, i) for i in xrange(10))().get()
[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]
# 组也支持部分参数化
g = group(add.s(i) for i in xrange(10))
g(10).get()
[10, 11, 12, 13, 14, 15, 16, 17, 18, 19]
Chains
把任务连接起来，执行完一个后把结果传入另一个再调用。
chain(add.s(4, 4) | mul.s(8))().get()
g = chain(add.s(4) | mul.s(8))
g(4).get()
# 也支持部分参数化
chords
chord是一个带callback的group
from celery import chord
from proj.tasks import add, xsum
chord((add.s(i, i) for i in xrange(10)), xsum.s())().get()
# 而group的结果发送到另一个任务则自动转换为chord
(group(add.s(i, i) for i in xrange(10)) | xsum.s())().get()
# 可以把单任务的结果发送到组中
upload_doc.s(file) | group(apply_filter.s() for filter in filters)
Routing
在app层面配置task到某个队列：
app.conf.update(
task_routes = {
'proj.tasks.add': {'queue': 'hipri'},
},
)
在调用时指定使用某个队列：
add.apply_async((2, 2), queue='hipri')
启动工作进程时指定该进程处理的队列：
celery -A proj worker -Q hipri
# 指定多个处理的队列 默认队列名称：celery
celery -A proj worker -Q hipri,celery
远程控制
查看工作进程当前的工作：
celery -A proj inspect active
这个命令是一个广播，所有工作进程都会收到。（需要启用event）
celery -A proj inspect active --destination=celery@example.com
让工作进程启用event
celery -A proj control enable_events