Celery是一款非常简单、灵活、可靠的分布式系统，可用于处理大量消息，并且提供了一整套操作此系统的工具。Celery 也是一款消息队列工具，可用于处理实时数据以及任务调度。
本文是是celery源码解析的第六 篇，在前五篇里分别介绍了:
神器 celery 源码解析- vine实现Promise功能 神器 celery 源码解析- py-amqp实现AMQP协议 神器 celery 源码解析- kombu，一个python实现的消息库 神器 celery 源码解析- kombu的企业级算法 神器 celery 源码解析- celery启动流程分析 本章我们跟着日志一起看看一次完整的任务调度流程，从另外一个角度了解启动过程中celery都做了什么。
worker模式启动流程 我们启动celery的worker, 启动大概分成3个阶段，先看第一阶段创建蓝图:
✗ celery -A myapp worker -l DEBUG这一阶段主要启动了worker和consumer2个蓝图, 下面是蓝图的创建和日志可以完整对应:
class Blueprint:# 创建蓝图 'Preparing bootsteps.' )'Building graph...' )for S in self._finalize_steps(steps):'New boot order: {%s}' ,', ' .join(s.alias for s in self.order))for step in order:return self第一个Worker蓝图在WorkController中，包括了下面一些步骤:
class WorkController:"" "Worker bootstep blueprint." "" 'Worker' 'celery.worker.components:Hub' ,'celery.worker.components:Pool' ,'celery.worker.components:Beat' ,'celery.worker.components:Timer' ,'celery.worker.components:StateDB' ,'celery.worker.components:Consumer' ,'celery.worker.autoscale:WorkerComponent' ,第二个Consumer蓝图在Consumer中，包括了下面一些步骤:
class Consumer:"" "Consumer blueprint." "" "" "Consumer blueprint." "" 'Consumer' 'celery.worker.consumer.connection:Connection' ,
,'celery.worker.consumer.events:Events' ,'celery.worker.consumer.gossip:Gossip' ,'celery.worker.consumer.heart:Heart' ,'celery.worker.consumer.control:Control' ,'celery.worker.consumer.tasks:Tasks' ,'celery.worker.consumer.consumer:Evloop' ,'celery.worker.consumer.agent:Agent' ,创建完2个蓝图后，并没有立即启动蓝图，转而进入第二阶段创建启动worker，日志输出如下:
...enable -E to monitor tasks in this worker)这个过程app创建完成，把当前的配置信息，task列表都展示出来。展示信息的模版:
BANNER = "" "\ "" "" " "" task信息来自app的tasks，在上篇我们介绍过，其实就是TaskRegistry；并发模式默认使用的prefork，多进程模式；然后是AMQP的消费者，queue，exchange等信息:
def extra_info(self):if self.loglevel <= logging.INFO:' . {task}' for task in sorted(self.app.tasks)if (not task.startswith(int_) if not include_builtins else task)return EXTRA_INFO_FMT.format(tasks=tasklist)'{}:{:#x}' .format(app.main or '__main__' , id(app))我们可以查看celery的进程数，确认总共创建了12个进程(进程数是通过cpu核数计算出来):
➜ ~ ps -ef | grep celery除了默认的多进程方式，celery还支持下面这些并发模式:
ALIASES = {'prefork' : 'celery.concurrency.prefork:TaskPool' ,'eventlet' : 'celery.concurrency.eventlet:TaskPool' ,'gevent' : 'celery.concurrency.gevent:TaskPool' ,'solo' : 'celery.concurrency.solo:TaskPool' ,'processes' : 'celery.concurrency.prefork:TaskPool' , # XXX compat alias 'threads' :
"" "Return pool implementation by name." "" return symbol_by_name(cls, ALIASES)threads 需要concurrent.futures支持，也就是python3.2版本以上
worker启动的第3阶段就是启动蓝图，日志如下:
[2021-11-24 15:53:13,062: DEBUG/MainProcess] | Worker: Starting Hubfor neighbors在worker启动中，我们需要关注worker蓝图的hub，pool二步(step)，consumer蓝图的connection，events，mingle，task，control，gossip，heart和Evloop七步(step)。
beat模式启动流程 beat模式的启动和worker模式不一样。beat模式主要是定时处理，并且beat模式不执行具体的任务，只是负责触发定时任务。其启动日志如下:
✗ celery -A myapp beat -l DEBUGin 5.00 minutes.in 5.00 minutes.从日志可以看到beat模式启动也大概可以分成2个阶段。第一个阶段就是创建和启动任务调度器，由beat命令提供:
class Beat:"" "Beat as a service." "" print (str(self.colored.cyan('celery beat v{VERSION_BANNER} is starting.' )))第二个阶段，任务调度器开始时间循环:
# celery/beat.py "" "Celery periodic task service." "" 'beat: Starting...' )'beat: Ticking with max interval->%s' ,if embedded_process:'celery beat' )while not self._is_shutdown.is_set():if interval and interval > 0.0:'beat: Waking up %s.' ,'in ' ))if self.scheduler.should_sync():这里的时间循环使用一个while循环去完成，每次tick都会检查是否有需要执行的任务，默认5分钟检查一次。
如果到达任务执行的时刻，则是通过下面的apply_async发送到worker(远程)去执行:
def apply_async(self, entry, producer=None, advance=True, **kwargs):# Update time-stamps and run counts before we actually execute, # so we have that done if an exception is raised (doesn't schedule # forever.) if advance else entryif isinstance(v, BeatLazyFunc) else v for v in (entry.args or [])]if isinstance(v, BeatLazyFunc) else v for k, v in entry.kwargs.items()}return task.apply_async(entry_args, entry_kwargs,multi模式启动流程 使用multi模式启动celery，可以让celery以服务的形式在background执行任务，并且可以启动更多的celery的执行进程。使用下面命令启动2个node ，w1和w2。
✗ celery multi start w1 w2 -A myapp -l DEBUG注意这个命令需要sudo权限
使用下面命令监测celery服务的状态。
✗ celery -A myapp statusw1的启动流程会写入到日志，日志内容如下:
✗ cat /var/log /celery/w1.logfor neighbors'w2@bogon' , revoked={}) [reply_to:{'exchange' : 'reply.celery.pidbox' , 'routing_key' : '196c0b68-a329-3e09-a1cf-54abb5e057db' } ticket:e640e757-9514-436c-8548-0ddcbe15f9a4]w1的启动方式和worker模式基本一致，特别的地方在日志的最后部分显示w2启动完成后，w1和w2进行了互联。对应可以在w2的日志中看到w1的连接信息:
✗ cat /var/log /celery/w2.log所以multi模式的特点就是新增加了Cluster和Node的概念，用来管理所有的worker，主要代码如下:
@splash'> Starting nodes...' )return int(any(cluster.start()))return [self.start_node(node) for node in self]return retcodeNode直接同步是在Gossip的step中:
class Gossip(bootsteps.ConsumerStep):'%s joined the party' , worker.hostname)完成测试后，可以使用命令 celery multi stop w1 w2 关闭node
worker接收任务流程 worker接收任务并执行的日志如下:
[2021-11-24 21:33:50,535: INFO/MainProcess] Received task: myapp.add[e9bb4aa0-8280-443f-a5ed-3deb0a0b99c2]function _trace_task_ret at 0x7fe6086ac280> (args:('myapp.add' , 'e9bb4aa0-8280-443f-a5ed-3deb0a0b99c2' , {'lang' : 'py' , 'task' : 'myapp.add' , 'id' : 'e9bb4aa0-8280-443f-a5ed-3deb0a0b99c2' , 'shadow' : None, 'eta' : None, 'expires' : None, 'group' : None, 'group_index' : None, 'retries' : 0, 'timelimit' : [None, None], 'root_id' : 'e9bb4aa0-8280-443f-a5ed-3deb0a0b99c2' , 'parent_id' : None, 'argsrepr' : '(16, 16)' , 'kwargsrepr' : '{}' , 'origin' : 'gen83110@192.168.5.28' , 'reply_to' : '63862dbb-9d82-3bdd-b7fb-03580941362a' , 'correlation_id' : 'e9bb4aa0-8280-443f-a5ed-3deb0a0b99c2' , 'hostname' : 'celery@192.168.5.28' , 'delivery_info' : {'exchange' : '' , 'routing_key' : 'celery' , 'priority' : 0, 'redelivered' : None}, 'args' : [16, 16], 'kwargs' : {}}, b'[[16, 16], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]' , 'application/json' , 'utf-8' ) kwargs:{})in 0.000271957000000711s: 32从日志信息可以看到，主进程MainProcess收到task执行的请求，然后从任务池中获取到任务，然后调度任务到一个子进程ForkPoolWorker-9中执行。
# celery/worker/strategy.py "" "Default task execution strategy. "" 'Received task: %s' , req)任务池是由并发模型提供:
# celery/concurrency/base.py "" "Equivalent of the :func:`apply` built-in function. "" if not kwargs else kwargsif not args else argsif self._does_debug:'TaskPool: Apply %s (args:%s kwargs:%s)' ,return self.on_apply(target, args, kwargs,小结 我们通过对worker，beat和multi三种启动模式的日志跟踪分析，对celery的启动流程和模块功能有更进一步的了解。
三个模式都需要创建app，所以启动时候通过参数-A myapp参数，由app创建/查找各种task。不同的地方首先是beat和worker/multi不同，beat实际上就是一个生产者，通过配置定时的产生任务，然后发送给worker/multi具体执行。
其次不同的是worker和multi的运作方式，multi以服务方式运行，并且可以跨机器。在worker模式下，本机创建多个工作进程，是一个多进程模型。multi则是多个机器Node形成一个Cluster集群，任务在集群内部进行调度。
celery的分布式模型大概可以如下图:
同时通过运行日志分析，我们可以知道celery的启动过程通过不同的Blueprint的不同Step过程实现；定时功能主要在beat和schedule模块实现；而分布式功能主要在concurrency模块，这样对各个模块的主体功能分工会有更清晰的认知。
Python猫技术交流群开放啦！ 交流群 （谢绝广告党，非诚勿扰！） ~
