目录
Celery是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列,同时也支持任务调度。
前面我们用几篇文章分析了 Kombu,为 Celery 的分析打下了基础。
本系列将继续通过源码分析,和大家一起深入学习 Celery。本文是系列第一篇,借鉴了几位网友的大作,按照自己的理解再重新整理,遂得此文。
Celery是Python世界中最受欢迎的后台工作管理者之一。它是一个简单、灵活且可靠的,处理大量消息的分布式系统,专注于实时处理的异步任务队列,同时也支持任务调度。
利用多线程,如Eventlet,gevent等,Celery的任务能被并发地执行在单个或多个工作服务器(worker servers)上。任务能异步执行(后台运行)或同步执行(等待任务完成)。Celery用于生产系统时候每天可以处理数以百万计的任务。
Celery是用Python编写的,但该协议可以在任何语言实现。它也可以与其他语言通过webhooks实现。
Celery建议的消息队列是RabbitMQ,但也支持Redis, Beanstalk, MongoDB, CouchDB, 和数据库(使用SQLAlchemy的或Django的 ORM) 。并且可以同时充当生产者和消费者。
使用Celery的常见场景如下:
Celery提供了如下的特性:
消息队列和任务队列,最大的不同之处就在于理念的不同 -- 消息队列传递的是“消息”,任务队列传递的是“任务”。
Celery 的基本逻辑为:分布式异步消息任务队列。
在 Celery 中,采用的是分布式的管理方式,每个节点之间都是通过广播/单播进行通信,从而达到协同效果。实际上,只有部分辅助管理功能才会协同,基础业务功能反而没有借助协同。
Celery包含如下组件:
再理解一下:
Celery 通过消息机制进行通信,通常使用中间人(Broker)作为客户端和职程(Worker)调节。启动一个任务的流程是:
Celery的架构图如下所示:
+-----------+ +--------------+
| Producer | | Celery Beat |
+-------+---+ +----+---------+
| |
| |
v v
+-------------------------+
| Broker |
+------------+------------+
|
|
|
+-------------------------------+
| | |
v v v
+----+-----+ +----+------+ +-----+----+
| Exchange | | Exchange | | Exchange |
+----+-----+ +----+------+ +----+-----+
| | |
v v v
+-----+ +-------+ +-------+
|queue| | queue | | queue |
+--+--+ +---+---+ +---+---+
| | |
| | |
v v v
+---------+ +--------+ +----------+
| worker | | Worker | | Worker |
+-----+---+ +---+----+ +----+-----+
| | |
| | |
+-----------------------------+
|
|
v
+---+-----+
| backend |
+---------+
目前我们得到如下信息:
下面我们就需要依据 Kombu来推论 Celery 应该如何设计。
首先,我们看看为了完成基本功能,Celery 应该具备哪些组件(模块),我们会提出一些问题,这些问题将在后续的分析中陆续得到解答。
因为Celery 的基本逻辑为:分布式异步消息任务队列,所以Celery包含如下基础组件:
以上为基础功能,但是作为分布式异步消息任务队列,我们还需要辅助功能(以及相关问题),比如。
进一步问题是:这些辅助功能是作为基础功能模块的一部分?还是独立出来成为一个功能模块?
这其实是一个哲学问题,每种实现都有其道理,或者说,很多决定其实就是作者灵光一现(临时拍脑袋)的产物。
比如我们后面提到的 Consumer 组件,表面上看,就是一个从broker获取消息的功能模块,直接使用 kombu 的 consumer 就可以做到。
但是实际上,celery Consumer 组件的概念远远要大于Kombu的Consumer,不只是利用了Kombu的Consumer从broker取得消息。也包括消息的消费,分发,监控,心跳等一系列功能。可以说,除了消息循环引擎 被 hub 承担,多进程被 Pool,Autoscaler 承担,定时任务被 timer,beat 承担之外,其他主要功能都被 Consumer 承担。
因此,我们需要看看:
Celery如果想成为消息处理系统,首先需要解决消息协议和消息传输问题。
所以我们首先看看如何封装 AMQP / Kombu。
具体封装是在 celery/app/amqp.py 文件中,其中主要有两个类:AMQP 和 Queues。
AMQP类的功能是 发送/接受消息,是对amqp协议实现的再一次封装,在这里其实就是对 kombu 类的再一次封装。
我们可以看到,其内部成员变量都是来自于 Kombu。比如 Connection, Consumer, Exchange, Producer, Queue, pools。
from kombu import Connection, Consumer, Exchange, Producer, Queue, pools
class AMQP:
"""App AMQP API: app.amqp."""
Connection = Connection
Consumer = Consumer
Producer = Producer
#: compat alias to Connection
BrokerConnection = Connection
queues_cls = Queues
#: Cached and prepared routing table.
_rtable = None
#: Underlying producer pool instance automatically
#: set by the :attr:`producer_pool`.
_producer_pool = None
# Exchange class/function used when defining automatic queues.
# For example, you can use ``autoexchange = lambda n: None`` to use the
# AMQP default exchange: a shortcut to bypass routing
# and instead send directly to the queue named in the routing key.
autoexchange = None
为了更好的理解,我们打印出amqp类的具体内容来看看。
amqp = {AMQP}
BrokerConnection = {type} <class 'kombu.connection.Connection'>
Connection = {type} <class 'kombu.connection.Connection'>
Consumer = {type} <class 'kombu.messaging.Consumer'>
Producer = {type} <class 'kombu.messaging.Producer'>
app = {Celery} <Celery myTest at 0x252bd2903c8>
autoexchange = {NoneType} None
default_exchange = {Exchange} Exchange celery(direct)
default_queue = {Queue} <unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>
producer_pool = {ProducerPool} <kombu.pools.ProducerPool object at 0x00000252BDC8F408>
publisher_pool = {ProducerPool} <kombu.pools.ProducerPool object at 0x00000252BDC8F408>
queues = {Queues: 1} {'celery': <unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>}
queues_cls = {type} <class 'celery.app.amqp.Queues'>
router = {Router} <celery.app.routes.Router object at 0x00000252BDC6B248>
routes = {tuple: 0} ()
task_protocols = {dict: 2} {1: <bound method AMQP.as_task_v1 of <celery.app.amqp.AMQP object at 0x00000252BDC74148>>, 2: <bound method AMQP.as_task_v2 of <celery.app.amqp.AMQP object at 0x00000252BDC74148>>}
_event_dispatcher = {EventDispatcher} <celery.events.dispatcher.EventDispatcher object at 0x00000252BE750348>
_producer_pool = {ProducerPool} <kombu.pools.ProducerPool object at 0x00000252BDC8F408>
_rtable = {tuple: 0} ()
具体逻辑如下:
+---------+
| Celery | +----------------------------+
| | | celery.app.amqp.AMQP |
| | | |
| | | |
| | | BrokerConnection +-----> kombu.connection.Connection
| | | |
| amqp+----->+ Connection +-----> kombu.connection.Connection
| | | |
+---------+ | Consumer +-----> kombu.messaging.Consumer
| |
| Producer +-----> kombu.messaging.Producer
| |
| producer_pool +-----> kombu.pools.ProducerPool
| |
| queues +-----> celery.app.amqp.Queues
| |
| router +-----> celery.app.routes.Router
+----------------------------+
Queues 则是一个扩展,一个逻辑概念,可以认为是 Broker 概念的进一步缩减版。
Producer 把任务发送给 Queues,Worker 从 Queues 获取任务,进行消费。
app.amqp.queues 就是 Queues 的一个实例,在其中存储了本 Worker 可以读取的所有 kombu.Queue。
class Queues(dict):
"""Queue name⇒ declaration mapping.
Arguments:
queues (Iterable): Initial list/tuple or dict of queues.
create_missing (bool): By default any unknown queues will be
added automatically, but if this flag is disabled the occurrence
of unknown queues in `wanted` will raise :exc:`KeyError`.
max_priority (int): Default x-max-priority for queues with none set.
"""
#: If set, this is a subset of queues to consume from.
#: The rest of the queues are then used for routing only.
_consume_from = None
def __init__(self, queues=None, default_exchange=None,
create_missing=True, autoexchange=None,
max_priority=None, default_routing_key=None):
dict.__init__(self)
self.aliases = WeakValueDictionary()
self.default_exchange = default_exchange
self.default_routing_key = default_routing_key
self.create_missing = create_missing
self.autoexchange = Exchange if autoexchange is None else autoexchange
self.max_priority = max_priority
if queues is not None and not isinstance(queues, Mapping):
queues = {q.name: q for q in queues}
queues = queues or {}
for name, q in queues.items():
self.add(q) if isinstance(q, Queue) else self.add_compat(name, **q)
对于一个 Consumer,可以配置其 queue,一个 Consumer 可以有多个queue,比如:
def add_consumer(state, queue, exchange=None, exchange_type=None,
routing_key=None, **options):
"""Tell worker(s) to consume from task queue by name."""
state.consumer.call_soon(
state.consumer.add_task_queue,
queue, exchange, exchange_type or 'direct', routing_key, **options)
return ok(f'add consumer {queue}')
add_consumer 名字个人认为有一定误导,其实是添加 queue,但是名字看起来像添加 Consumer。
而在 Consumer 之中,会对 queues 进行具体配置。
def add_task_queue(self, queue, exchange=None, exchange_type=None,
routing_key=None, **options):
cset = self.task_consumer
queues = self.app.amqp.queues
if queue in queues:
q = queues[queue]
else:
exchange = queue if exchange is None else exchange
exchange_type = ('direct' if exchange_type is None
else exchange_type)
q = queues.select_add(queue,
exchange=exchange,
exchange_type=exchange_type,
routing_key=routing_key, **options)
if not cset.consuming_from(queue):
cset.add_queue(q)
cset.consume()
info('Started consuming from %s', queue)
通过以上的分析,大家应该对 Celery 的架构有了初步的了解。在下篇文章中,我们将从几个方面做进一步思考,敬请期待。