celery任务状态监控
celery是python里常用的一个异步任务队列，使用celery可以大大简化搭建任务队列的工作。实际使用中可能会需要监控一些任务或者定时任务的运行状态。
这里就讲一下celery的任务状态监控相关的方法。
单独使用celery命令格式为 celery -A [proj] [cmd]
在django下使用时，用manage.py启动时则不需要-A参数，命令格式为
python manage.py celery [cmd]
1. 通用命令
1） 查看celery worker的状态
python manage.py status
celery@DEV_192_168_X_XX: OK
celery@DEV_192_168_X_XX: OK
celery@DEV_192_168_X_XX: OK
celery@DEV_192_168_X_XX: OK
celery@DEV_192_168_X_XX: OK
1 node online.
返回内容表示目前有5个worker运行中，运行正常，
@前面是worker的name，后面是worker的hostname，在启动worker时用-n workername@%h参数传入，如果不传递，默认为celery@hostname。
会报一个warning，建议给每个worker定义一个唯一的名字。
第二部分返回值是worker的计数，只是简单的统计了不同名称的个数.比如上面没有指定队列名称，就会统计只有一个活动的node.
2） 查看 celery task和queue详细信息
查看当前正在运行的task
python manage.py inspect active
-> celery@localhost: OK
- empty -
查看当前的队列的具体信息
python manage.py inspect active-queues
-> celery@localhost: OK
* {u'exclusive': False, u'name': u'celery', u'exchange': {u'name': u'celery'
, u'durable': True, u'delivery_mode': 2, u'passive': False, u'arguments': None,
u'type': u'direct', u'auto_delete': False}, u'durable': True, u'routing_key': u'
celery', u'no_ack': False, u'alias': None, u'queue_arguments': None, u'binding_a
rguments': None, u'bindings': [], u'auto_delete': False}
输出 worker的内存占用(需要安装psutil库)
python manage.py celery inspect memdump
-> celery@localhost: OK
- rss (end): 37.352MB.
输出worker的统计信息
python manage.py celery inspect stats
-> celery@localhost: OK
{
"broker": {
"alternates": [],
"connect_timeout": 4,
"failover_strategy": "round-robin",
"heartbeat": 0,
"hostname": "192.168.99.100",
"insist": false,
"login_method": null,
"port": 6379,
"ssl": false,
"transport": "redis",
"transport_options": {},
"uri_prefix": null,
"userid": null,
"virtual_host": "1"
},
"clock": "7624",
"pid": 8308,
"pool": {
"max-concurrency": 4,
"max-tasks-per-child": "N/A",
"processes": [
8584,
9180,
9792,
5600
],
"put-guarded-by-semaphore": true,
"timeouts": [
0,
0
],
"writes": "N/A"
},
"prefetch_count": 16,
"rusage": "N/A",
"total": {
"tasks.jobs.test": 1562
}
}
broker段记录了连接的broker信息，pool段有当前子进程/子线程号，total则是当前task运行到现在记录的所有数据总和。这些信息在worker退出后会被重置。
如果worker进程已经异常退出，则无法用inspect获取信息统计。
字段更多详细内容可查阅以下文档
http://docs.celeryproject.org/en/latest/userguide/workers.html#worker-statistics
Inspect还有很多其他参数，可以输入 celery inspect --help 来查看。
主要常用的就是以上几条命令.
3） 清空队列
当消息堆积，或者worker挂掉，但是队列已经塞入了大量无用信息时。如果不关注堆积消息的处理（比如一些定时重复计算的任务）。可以直接清空队列。
防止启动work后，需要处理冗长的堆积消息而导致后续消息处理延迟。
python manage.py celery purge -Q [queue_name]
建议带上-Q参数，清空指定队列。
旧版本的celery不支持-Q参数，可以使用对应的broke指令，清理一下对应的队列。可以参考下文中redis清除队列，以及rabbitmq purge队列的指令
4） 使用flower监控task运行结果和状态。
直接安装flower
pip install flower
启动flower，开启持久化，如果不带persisitent参数。重启flower就会导致数据丢失
python manage.py flower --port=5555 --persistent=True --db=./flower
Flower的除了界面样式优点陈旧之外其他都很好用，另外可以使用参数开启权限认证。不过界面上有停止worker的操作和调整work的参数，可能会导致一些安全性的问题。内网监控可以使用。
2. 查看消息队列内部内容
1） 使用redis作为Broker
需要先安装redis环境，使用redis-cli查看。
基本就是使用redis指令对存储在redis里的list进行各种操作。
查看当前所有队列
即查看当前存储的所有的key
redis-cli -h HOST -p PORT -n DB_NUMBER(默认为0) keys *
# redis-cli -n 1 keys \*
1) "_kombu.binding.celery.pidbox"
2) "_kombu.binding.celeryev"
3) "_kombu.binding.celery"
_kombu.binding.celery是当前的工作队列， pidbox和celeryev是celery内部用来发送event和管理其他队列的队列。
清空队列
# 清空队列全部内容
redis-cli -h HOST -p PORT -n DB_NUMBER ltrim key 0 0
# 清除前100条消息
redis-cli -h HOST -p PORT -n DB_NUMBER ltrim key 100 -1
# 保留前100条消息
redis-cli -h HOST -p PORT -n DB_NUMBER ltrim key 0 100
查看消息长度
redis-cli -h HOST -p PORT -n DB_NUMBER llen [上面列出的key的最后一段]
# redis-cli -n 1 llen celery
(integer) 9
查看消息队列具体内容
redis-cli -h HOST -p PORT -n DB_NUMBER LRANGE key 0 -1
# redis-cli -n 1 LRANGE celery 0 -1
17) "{\"body\": \"eyJleHBpcmVzIjogbnVsbCwgInV0YyI6IHRydWUsICJhcmdzIjogW10sICJjaG
9yZCI6IG51bGwsICJjYWxsYmFja3MiOiBudWxsLCAiZXJyYmFja3MiOiBudWxsLCAidGFza3NldCI6IG
51bGwsICJpZCI6ICI2OGQyZGJlYS1jNjdiLTRmMmYtOGQ1My02NDQ5M2JkMzlmNmQiLCAicmV0cmllcy
I6IDAsICJ0YXNrIjogInRhc2tzLmpvYnMudGVzdCIsICJ0aW1lbGltaXQiOiBbbnVsbCwgbnVsbF0sIC
JldGEiOiBudWxsLCAia3dhcmdzIjoge319\", \"headers\": {}, \"content-type\": \"appli
cation/json\", \"properties\": {\"body_encoding\": \"base64\", \"correlation_id\
": \"68d2dbea-c67b-4f2f-8d53-64493bd39f6d\", \"reply_to\": \"42f2e00e-cd9a-3259-
a201-d5e94c869b90\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"cel
ery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"delivery_tag\": \"d93d
25ee-dc33-486a-88ac-af61b123b615\"}, \"content-encoding\": \"utf-8\"}"
18) "{\"body\": \"eyJleHBpcmVzIjogbnVsbCwgInV0YyI6IHRydWUsICJhcmdzIjogW10sICJjaG
9yZCI6IG51bGwsICJjYWxsYmFja3MiOiBudWxsLCAiZXJyYmFja3MiOiBudWxsLCAidGFza3NldCI6IG
51bGwsICJpZCI6ICI4MTk5Y2RiMy0wNjNmLTQ1ZTYtYTIwOS1kNWE5NjNmN2ZiYzIiLCAicmV0cmllcy
I6IDAsICJ0YXNrIjogInRhc2tzLmpvYnMudGVzdCIsICJ0aW1lbGltaXQiOiBbbnVsbCwgbnVsbF0sIC
JldGEiOiBudWxsLCAia3dhcmdzIjoge319\", \"headers\": {}, \"content-type\": \"appli
cation/json\", \"properties\": {\"body_encoding\": \"base64\", \"correlation_id\
": \"8199cdb3-063f-45e6-a209-d5a963f7fbc2\", \"reply_to\": \"42f2e00e-cd9a-3259-
a201-d5e94c869b90\", \"delivery_info\": {\"priority\": 0, \"routing_key\": \"cel
ery\", \"exchange\": \"celery\"}, \"delivery_mode\": 2, \"delivery_tag\": \"2246
c451-3221-4579-b43b-ed5fa8085661\"}, \"content-encoding\": \"utf-8\"}"
大致可以看出消息被编码成了base64的格式，还有各个信息的routing_key和exchange，关于消息的编码，各位有兴趣可以继续深究。
2 ）使用Rabbit-mq作为Broker
需要rabbitmq环境，因为rmq本身就是消息队列，所以每个队列对应于rmq里的一个queue，具体操作和管理rabbitmq相同。
队列状态
sudo rabbitmqctl list_queues
sudo rabbitmqctl list_queues name messages messages_ready messages_unacknowledged
sudo rabbitmqctl list_queues name consumers
sudo rabbitmqctl list_queues name memory
队列消息查看
需要使用python的工具来辅助管理，参考文档
http://www.rabbitmq.com/management-cli.html
sudo python rabbitmqadmin get queue=queuename requeue=true count=10
3. 监控celery的事件
注意：如果work里面设置event=disable，就无法监听。
直接参考以下代码进行监控
from celery import Celery
def my_monitor(app):
state = app.events.State()
def announce_failed_tasks(event):
state.event(event)
# task name is sent only with -received event, and state
# will keep track of this for us.
task = state.tasks.get(event['uuid'])
print('TASK FAILED: %s[%s] %s' % (
task.name, task.uuid, task.info(),))
def announce_receive_tasks(event):
state.event(event)
# task name is sent only with -received event, and state
# will keep track of this for us.
task = state.tasks.get(event['uuid'])
print('TASK RECEIVED: %s[%s] %s' % (
task.name, task.uuid, task.info(),))
with app.connection() as connection:
recv = app.events.Receiver(connection, handlers={
'task-failed': announce_failed_tasks,
'task-received': announce_receive_tasks,
'*': state.event,
})
recv.capture(limit=None, timeout=None, wakeup=True)
if __name__ == '__main__':
print 'monitor start'
app = Celery(broker='redis://127.0.0.1:6379/1')
my_monitor(app)
可以监听到的所有事件在这里
http://docs.celeryproject.org/en/latest/userguide/monitoring.html#event-reference
4. 其他补充
Celery本身运行的worker进程直接监控运行中的进程状态即可。
比如使用supervisor管理进程。这里不再赘述
