监控Celery不一定非要使用Flower
运维平台中有许多的周期/定时/异步任务，例如证书扫描、数据备份、日志清理、线上作业等等，这些任务的执行都是借助于Celery来完成的。任务多了之后就会遇到一系列的问题，例如我之前写过的
将任务分多队列来解决生产环境下的任务优先级问题
，除此之外还要经常关注队列的状态以及任务的运行情况，为了方便查看任务以及Worker的运行情况，我在后台中添加了队列状态这个功能
这个功能参考了Celery知名监控工具Flower的实现，之所以没有直接使用Flower，主要有几方面的考虑，其一是Flower的页面风格与我们本身的平台风格差异较大，其二是Flower要单独启动进程或者是去看代码深度集成，比较麻烦，其三也是想要更加深入的了解下Celery，毕竟项目用了它，了解不深遇到问题不好解决。基于以上几点考虑，就自己实现了Celery监控功能的开发。好在并不复杂，Celery本身提供了一组API，可以查询任务队列的状态、执行结果等信息。借助于这些API就能完成比较详细的监控，Celery的API主要有两个
inspect
app.control.inspect(): 这个方法返回一个Inspect对象，可以使用它来获取任务队列、工作节点等的信息。例如
inspect().active()
可以获取当前活动的任务列表，
inspect().registered()
可以获取已注册的任务列表，不指定worker的情况下查看全部worker的数据，如果指定worker则查看对应worker的数据
具体的用法如下：
from celery import Celery
app = Celery('your_celery_app_name')
# 检查工作节点的在线状态
worker_status = app.control.inspect([worker]).ping()
# 返回工作节点的统计信息，如活动任务数、完成任务数等
worker_stats = app.control.inspect([worker]).stats()
# 返回活动任务的信息
active_tasks = app.control.inspect([worker]).active()
# 返回已注册任务的信息
registered_tasks = app.control.inspect([worker]).registered()
# 返回计划中的任务的信息
scheduled_tasks = app.control.inspect([worker]).scheduled()
# 返回已预订任务的信息
reserved_tasks = app.control.inspect([worker]).reserved()
# 返回已撤销任务的信息
revoked_tasks = app.control.inspect([worker]).revoked()
# 返回活动队列的信息
active_queues = app.control.inspect([worker]).active_queues()
# 查询worker的配置信息
worker_conf = app.control.inspect([worker]).conf()
# 返回工作节点的报告信息
worker_reports = app.control.inspect([worker]).report()
# 查询特定任务的信息
task_info = app.control.inspect([worker]).query_task(task_id)
通过
inspect
可以查看celery整体以及单个worker的相关信息，例如首页的状态就是通过
ping
来实现的
一个card表示一个worker，点击card可以查看worker的详细信息，例如Pool标签的数据就来自于
stats
，Register标签的数据就来自于
registered
，Tasks标签就分别展示了
active
、
scheduled
、
reserved
和
revoked
的任务数据
通过这些信息可以全面掌握Celery的运行状态以及每个Worker的运行情况，做到对Celery的全局掌控
AsyncResult
AsyncResult: 这个类可以查询特定任务的状态和结果。通过任务的ID创建一个AsyncResult对象，并使用其方法来获取任务的状态、结果等信息。例如
AsyncResult(task_id).state
可以获取任务的状态，
AsyncResult(task_id).result
可以获取任务的执行结果
具体的用法如下：
from celery import Celery
app = Celery('your_celery_app_name')
result = app.AsyncResult(task_id)
# 获取任务状态
state = result.state
# 获取任务结果
result = result.result
# 返回一个布尔值，检查任务是否已经完成
is_ready = result.ready()
# 返回一个布尔值，检查任务是否成功完成
is_successful = result.successful()
# 返回一个布尔值，检查任务是否执行失败
is_failed = result.failed()
# 返回一个字符串，获取任务的错误追溯信息
traceback = result.traceback
# 返回一个AsyncResult对象，获取任务的父任务
parent_task = result.parent
# 返回一个列表，包含任务的子任务的AsyncResult对象，获取任务的子任务
child_tasks = result.children
# 返回一个字典，获取任务的其他信息
info = result.info
# 获取任务的结果，可以指定超时时间和是否向上传播异常
result = result.get(timeout=10, propagate=False)
# 忘记任务，将任务从结果存储中删除。一旦任务被遗忘，将无法查询其状态和结果
result.forget()
通过
AsyncResult
可以获取到任务执行的相关信息，对任务执行过程和结果都有很好的把控。不过这需要任务的ID，任务ID通常可以通过任务执行时获取，
ops_coffee_auto_notify_task.delay()
异步执行任务后返回的就是任务的ID。但对于任务的话我通常会单独记录任务执行的过程和状态，而不依赖Celery的结果记录，所以对于
AsyncResult
的需求并不强
但平常需要开关系统内置的周期任务，此时就需要知道系统任务列表，通过获取
PeriodicTask
表的数据即可，同时通过修改
PeriodicTask
表的
enabled
字段值来达到开启或暂停的目的
inspect
更关注Celery的Worker运行情况，而
AsyncResult
则更关注于Celery的任务运行状态，通过两者的配合就能更好的掌握周期/定时/异步任务的执行情况了，通过监控功能的开发也对Celery有了更深的了解，同时
运维自动化平台
也变得更加完善