[源码分析] 分布式任务队列 Celery 之发送Task & AMQP

Posted 2021-04-23 罗西的思考

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了[源码分析] 分布式任务队列 Celery 之发送Task & AMQP相关的知识，希望对你有一定的参考价值。

[源码分析] 分布式任务队列 Celery 之发送Task & AMQP

0x00 摘要0x01 示例代码1.1 服务端1.2 客户端0x02 系统启动2.1 产生Celery2.2 task 装饰器2.2.1 添加任务2.2.2 绑定2.3 小结0x03 amqp类3.1 生成3.2 定义0x04 发送Task4.1 apply_async in task4.2 send_task4.3 生成消息内容4.4 send_task_message in amqp4.5 publish in producer4.6 redis 内容0xFF 参考

0x00 摘要

Celery是一个简单、灵活且可靠的，处理大量消息的分布式系统，专注于实时处理的异步任务队列，同时也支持任务调度。

在之前的文章中，我们看到了关于Task的分析，本文我们重点看看在客户端如何发送Task，以及 Celery 的amqp对象如何使用。

在阅读之前，我们依然要提出几个问题，以此作为阅读时候的指引：

客户端启动时候，Celery 应用和用户自定义 Task 是如何生成的？
Task 装饰器起到了什么作用？
发送 Task 时候，消息是如何组装的？
发送 Task 时候，采用什么媒介（模块）来发送？amqp？
Task 发送出去之后，在 Redis 之中如何存储？

说明：在整理文章时，发现漏发了一篇，从而会影响大家阅读思路，特此补上，请大家谅解。

[源码分析] 消息队列 Kombu 之 mailbox

[源码分析] 消息队列 Kombu 之 Hub

[源码分析] 消息队列 Kombu 之 Consumer

[源码分析] 消息队列 Kombu 之 Producer

[源码分析] 消息队列 Kombu 之启动过程

[源码解析] 消息队列 Kombu 之基本架构

[ 源码解析] 并行分布式框架 Celery 之架构 (1)

[ 源码解析] 并行分布式任务队列 Celery 之架构 (2)

[ 源码解析] 并行分布式框架 Celery 之 worker 启动 (1)

[源码解析] 并行分布式框架 Celery 之 worker 启动 (2)

[ 源码解析] 并行分布式任务队列 Celery 之启动 Consumer

[ 源码解析] 并行分布式任务队列 Celery 之 Task是什么

[[从源码学设计]celery 之发送Task & AMQP ]() 就是本文，从客户端角度讲解发送Task

[源码解析] 并行分布式任务队列 Celery 之消费动态流程下一篇文章从服务端角度讲解收到 Task 如何消费

[源码解析] 并行分布式任务队列 Celery 之多进程模型

0x01 示例代码

我们首先给出示例代码。

1.1 服务端

示例代码服务端如下，这里使用了装饰器来包装待执行任务。

from celery import Celery

app = Celery('myTest', broker='redis://localhost:6379')

@app.task
def add(x,y):
    return x+y

if __name__ == '__main__':
    app.worker_main(argv=['worker'])

1.2 客户端

客户端发送代码如下，就是调用 add Task 来做加法计算：

from myTest import add
re = add.apply_async((2,17))

我们开始具体介绍，以下均是客户端的执行序列。

0x02 系统启动

我们首先要介绍在客户端，Celery 系统和 task（实例）是如何启动的。

2.1 产生Celery

如下代码首先会执行 myTest 这个 Celery。

app = Celery('myTest', broker='redis://localhost:6379')

2.2 task 装饰器

Celery 使用了装饰器来包装待执行任务（因为各种语言的类似概念，在本文中可能会混用装饰器或者注解这两个术语）

@app.task
def add(x,y):
    return x+y

task这个装饰器具体执行其实就是返回 _create_task_cls 这个内部函数执行的结果。

这个函数返回一个Proxy，Proxy 在真正执行到的时候，会执行 _task_from_fun。

_task_from_fun 的作用是：将该task添加到全局变量中，即当调用 _task_from_fun 时会将该任务添加到app任务列表中，以此达到所有任务共享的目的。这样客户端才能知道这个 task。

    def task(self, *args, **opts):
        """Decorator to create a task class out of any callable. """
        if USING_EXECV and opts.get('lazy', True):
            from . import shared_task
            return shared_task(*args, lazy=False, **opts)

        def inner_create_task_cls(shared=True, filter=None, lazy=True, **opts):
            _filt = filter

            def _create_task_cls(fun):
                if shared:
                    def cons(app):
                        return app._task_from_fun(fun, **opts) # 将该task添加到全局变量中，当调用_task_from_fun时会将该任务添加到app任务列表中，以此达到所有任务共享的目的
                    cons.__name__ = fun.__name__
                    connect_on_app_finalize(cons)
                if not lazy or self.finalized:
                    ret = self._task_from_fun(fun, **opts)
                else:
                    # return a proxy object that evaluates on first use
                    ret = PromiseProxy(self._task_from_fun, (fun,), opts,
                                       __doc__=fun.__doc__)
                    self._pending.append(ret)
                if _filt:
                    return _filt(ret)
                return ret

            return _create_task_cls

        if len(args) == 1:
            if callable(args[0]):
                return inner_create_task_cls(**opts)(*args) #执行在这里
        return inner_create_task_cls(**opts)

我们具体分析下这个装饰器。

2.2.1 添加任务

在初始化过程中，为每个app添加该任务时，会调用到app._task_from_fun(fun, **options)。

具体作用是：

判断各种参数配置；
动态创建task；
将任务添加到_tasks任务中；
用task的bind方法绑定相关属性到该实例上；

代码如下：

    def _task_from_fun(self, fun, name=None, base=None, bind=False, **options):

        name = name or self.gen_task_name(fun.__name__, fun.__module__)         # 如果传入了名字则使用，否则就使用moudle name的形式
        base = base or self.Task                                                # 是否传入Task，否则用类自己的Task类 默认celery.app.task:Task

        if name not in self._tasks:                                             # 如果要加入的任务名称不再_tasks中
            run = fun if bind else staticmethod(fun)                            # 是否bind该方法是则直接使用该方法，否则就置为静态方法
            task = type(fun.__name__, (base,), dict({
                'app': self,                                                    # 动态创建Task类实例
                'name': name,                                                   # Task的name
                'run': run,                                                     # task的run方法
                '_decorated': True,                                             # 是否装饰
                '__doc__': fun.__doc__,
                '__module__': fun.__module__,
                '__header__': staticmethod(head_from_fun(fun, bound=bind)),
                '__wrapped__': run}, **options))()                              
            # for some reason __qualname__ cannot be set in type()
            # so we have to set it here.
            try:
                task.__qualname__ = fun.__qualname__                            
            except AttributeError:
                pass
            self._tasks[task.name] = task                                       # 将任务添加到_tasks任务中
            task.bind(self)  # connects task to this app                        # 调用task的bind方法绑定相关属性到该实例上

            add_autoretry_behaviour(task, **options)
        else:
            task = self._tasks[name]
        return task

2.2.2 绑定

bind方法的作用是：绑定相关属性到该实例上，因为只知道 task 名字或者代码是不够的，还需要在运行时候拿到 task 的实例。

@classmethod
def bind(cls, app):
    was_bound, cls.__bound__ = cls.__bound__, True
    cls._app = app                                          # 设置类的_app属性
    conf = app.conf                                         # 获取app的配置信息
    cls._exec_options = None  # clear option cache

    if cls.typing is None:
        cls.typing = app.strict_typing

    for attr_name, config_name in cls.from_config:          # 设置类中的默认值
        if getattr(cls, attr_name, None) is None:           # 如果获取该属性为空
            setattr(cls, attr_name, conf[config_name])      # 使用app配置中的默认值

    # decorate with annotations from config.
    if not was_bound:
        cls.annotate()

        from celery.utils.threads import LocalStack
        cls.request_stack = LocalStack()                    # 使用线程栈保存数据

    # PeriodicTask uses this to add itself to the PeriodicTask schedule.
    cls.on_bound(app)

    return app

2.3 小结

至此，在客户端（使用者方），Celery 应用已经启动，一个task实例也已经生成，其属性都被绑定在实例上。

0x03 amqp类

在客户端调用 apply_async 的时候，会调用 app.send_task 来具体发送任务，其中用到 amqp，所以我们首先讲讲 amqp 类。

3.1 生成

在 send_task 之中有如下代码，就是：

    def send_task(self, ....):
        """Send task by name.
        """
        parent = have_parent = None
        amqp = self.amqp # 此时生成

此时的 self 是 Celery 应用本身，具体内容我们打印出来看看，从下面我们可以看到 Celery 应用是什么样子。

self = {Celery} <Celery myTest at 0x1eeb5590488>
 AsyncResult = {type} <class 'celery.result.AsyncResult'>
 Beat = {type} <class 'celery.apps.beat.Beat'>
 GroupResult = {type} <class 'celery.result.GroupResult'>
 Pickler = {type} <class 'celery.app.utils.AppPickler'>
 ResultSet = {type} <class 'celery.result.ResultSet'>
 Task = {type} <class 'celery.app.task.Task'>
 WorkController = {type} <class 'celery.worker.worker.WorkController'>
 Worker = {type} <class 'celery.apps.worker.Worker'>
 amqp = {AMQP} <celery.app.amqp.AMQP object at 0x000001EEB5884188>
 amqp_cls = {str} 'celery.app.amqp:AMQP'
 backend = {DisabledBackend} <celery.backends.base.DisabledBackend object at 0x000001EEB584E248>
 clock = {LamportClock} 0
 control = {Control} <celery.app.control.Control object at 0x000001EEB57B37C8>
 events = {Events} <celery.app.events.Events object at 0x000001EEB56C7188>
 loader = {AppLoader} <celery.loaders.app.AppLoader object at 0x000001EEB5705408>
 main = {str} 'myTest'
 pool = {ConnectionPool} <kombu.connection.ConnectionPool object at 0x000001EEB57A9688>
 producer_pool = {ProducerPool} <kombu.pools.ProducerPool object at 0x000001EEB6297508>
 registry_cls = {type} <class 'celery.app.registry.TaskRegistry'>
 tasks = {TaskRegistry: 10} {'myTest.add': <@task: myTest.add of myTest at 0x1eeb5590488>, 'celery.accumulate': <@task: celery.accumulate of myTest at 0x1eeb5590488>, 'celery.chord_unlock': <@task: celery.chord_unlock of myTest at 0x1eeb5590488>, 'celery.chunks': <@task: celery.chunks of myTest at 0x1eeb5590488>, 'celery.backend_cleanup': <@task: celery.backend_cleanup of myTest at 0x1eeb5590488>, 'celery.group': <@task: celery.group of myTest at 0x1eeb5590488>, 'celery.map': <@task: celery.map of myTest at 0x1eeb5590488>, 'celery.chain': <@task: celery.chain of myTest at 0x1eeb5590488>, 'celery.starmap': <@task: celery.starmap of myTest at 0x1eeb5590488>, 'celery.chord': <@task: celery.chord of myTest at 0x1eeb5590488>}

堆栈为：

amqp, base.py:1205
__get__, objects.py:43
send_task, base.py:705
apply_async, task.py:565
<module>, myclient.py:4

为什么赋值语句就可以生成 amqp？是因为其被 cached_property 修饰。

使用 cached_property 修饰过的函数，就变成是对象的属性，该对象第一次引用该属性时，会调用函数，对象第二次引用该属性时就直接从词典中取了，即 Caches the return value of the get method on first call。

    @cached_property
    def amqp(self):
        """AMQP related functionality: :class:`~@amqp`."""
        return instantiate(self.amqp_cls, app=self)

3.2 定义

AMQP类就是对amqp协议实现的再一次封装，在这里其实就是对 kombu 类的再一次封装。

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>
 argsrepr_maxsize = {int} 1024
 autoexchange = {NoneType} None
 default_exchange = {Exchange} Exchange celery(direct)
 default_queue = {Queue} <unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>
 kwargsrepr_maxsize = {int} 1024
 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>>}
 utc = {bool} True
  _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
               +----------------------------+

0x04 发送Task

我们接着看看客户端如何发送task。

from myTest import add
re = add.apply_async((2,17))

总述下逻辑：

Producer 初始化过程完成了连接用的内容，比如调用self.connect方法，到预定的Transport类中连接载体，并初始化Chanel，self.chanel = self.connection；
调用 Message 封装消息；
Exchange 将 routing_key 转为 queue；
调用 amqp 发送消息；
Channel 负责最终消息发布；

我们下面详细解读下。

4.1 apply_async in task

这里重要的是两点：

如果是 task_always_eager，则产生一个 Kombu . producer；
否则，调用 amqp 来发送 task（我们主要看这里）；

缩减版代码如下：

    def apply_async(self, args=None, kwargs=None, task_id=None, producer=None,
                    link=None, link_error=None, shadow=None, **options):
        """Apply tasks asynchronously by sending a message.
        """

        preopts = self._get_exec_options()
        options = dict(preopts, **options) if options else preopts

        app = self._get_app()
        if app.conf.task_always_eager:
            # 获取 producer
            with app.producer_or_acquire(producer) as eager_producer:      
                serializer = options.get('serializer')
                body = args, kwargs
                content_type, content_encoding, data = serialization.dumps(
                    body, serializer,
                )
                args, kwargs = serialization.loads(
                    data, content_type, content_encoding,
                    accept=[content_type]
                )
            with denied_join_result():
                return self.apply(args, kwargs, task_id=task_id or uuid(),
                                  link=link, link_error=link_error, **options)
        else:
            return app.send_task( #调用到这里
                self.name, args, kwargs, task_id=task_id, producer=producer,
                link=link, link_error=link_error, result_cls=self.AsyncResult,
                shadow=shadow, task_type=self,
                **options
            )

此时如下：

         1  apply_async       +-------------------+
                              |                   |
User  +---------------------> | task: myTest.add  |
                              |                   |
                              +-------------------+

4.2 send_task

此函数作用是生成任务信息，调用amqp发送任务：

获取amqp实例；
设置任务id，如果没有传入则生成任务id；
生成路由值，如果没有则使用amqp的router；
生成route信息；
生成任务信息；
如果有连接则生成生产者；
发送任务消息；
生成异步任务实例；
返回结果；

具体如下：

def send_task(self, name, ...):
    """Send task by name.
    """
    parent = have_parent = None
    amqp = self.amqp                                                    # 获取amqp实例
    task_id = task_id or uuid()                                         # 设置任务id，如果没有传入则生成任务id
    producer = producer or publisher  # XXX compat                      # 生成这
    router = router or amqp.router                                      # 路由值，如果没有则使用amqp的router
    options = router.route(
        options, route_name or name, args, kwargs, task_type)           # 生成route信息

    message = amqp.create_task_message( # 生成任务信息
        task_id, name, args, kwargs, countdown, eta, group_id, group_index,
        expires, retries, chord,
        maybe_list(link), maybe_list(link_error),
        reply_to or self.thread_oid, time_limit, soft_time_limit,
        self.conf.task_send_sent_event,
        root_id, parent_id, shadow, chain,
        argsrepr=options.get('argsrepr'),
        kwargsrepr=options.get('kwargsrepr'),
    )

    if connection:
        producer = amqp.Producer(connection)                            # 如果有连接则生成生产者

    with self.producer_or_acquire(producer) as P:                       
        with P.connection._reraise_as_library_errors():
            self.backend.on_task_call(P, task_id)
            amqp.send_task_message(P, name, message, **options)         # 发送任务消息 

    result = (result_cls or self.AsyncResult)(task_id)                  # 生成异步任务实例
    if add_to_parent:
        if not have_parent:
            parent, have_parent = self.current_worker_task, True
        if parent:
            parent.add_trail(result)
    return result                                                       # 返回结果

此时如下：

         1  apply_async       +-------------------+
                              |                   |
User  +---------------------> | task: myTest.add  |
                              |                   |
                              +--------+----------+
                                       |
                                       |
                        2 send_task    |
                                       |
                                       v
                                +------+--------+
                                | Celery myTest |
                                |               |
                                +------+--------+
                                       |
                                       |
                  3 send_task_message  |
                                       |
                                       v
                               +-------+---------+
                               |      amqp       |
                               |                 |
                               |                 |
                               +-----------------+

4.3 生成消息内容

as_task_v2 会具体生成消息内容。大家可以看到如果实现一个消息，需要用到几个大部分：

headers，包括：task name, task id, expires, 等等；
消息类型和编码方式：content-encoding，content-type；
参数：这些就是 Celery 特有的，用来区分不同队列的，比如：exchange，routing_key 等等；
body : 就是消息体；

最终具体消息举例如下：

{
    "body": "W1syLCA4XSwge30sIHsiY2FsbGJhY2tzIjogbnVsbCwgImVycmJhY2tzIjogbnVsbCwgImNoYWluIjogbnVsbCwgImNob3JkIjogbnVsbH1d",
    "content-encoding": "utf-8",
    "content-type": "application/json",
    "headers": {
        "lang": "py",
        "task": "myTest.add",
        "id": "243aac4a-361b-4408-9e0c-856e2655b7b5",
        "shadow": null,
        "eta": null,
        "expires": null,
        "group": null,
        "group_index": null,
        "retries": 0,
        "timelimit": [null, null],
        "root_id": "243aac4a-361b-4408-9e0c-856e2655b7b5",
        "parent_id": null,
        "argsrepr": "(2, 8)",
        "kwargsrepr": "{}",
        "origin": "gen33652@DESKTOP-0GO3RPO"
    },
    "properties": {
        "correlation_id": "243aac4a-361b-4408-9e0c-856e2655b7b5",
        "reply_to": "b34fcf3d-da9a-3717-a76f-44b6a6362da1",
        "delivery_mode": 2,
        "delivery_info": {
            "exchange": "",
            "routing_key": "celery"
        },
        "priority": 0,
        "body_encoding": "base64",
        "delivery_tag": "fa1bc9c8-3709-4c02-9543-8d0fe3cf4e6c"
    }
}

具体代码如下，这里的 sent_event 是后续发送时候需要，并不体现在具体消息内容之中：

def as_task_v2(self, task_id, name, args=None, kwargs=None, ......):

    ......

    return task_message(
        headers={
            'lang': 'py',
            'task': name,
            'id': task_id,
            'shadow': shadow,
            'eta': eta,
            'expires': expires,
            'group': group_id,
            'group_index': group_index,
            'retries': retries,
            'timelimit': [time_limit, soft_time_limit],
            'root_id': root_id,
            'parent_id': parent_id,
            'argsrepr': argsrepr,
            'kwargsrepr': kwargsrepr,
            'origin': origin or anon_nodename()
        },
        properties={
            'correlation_id': task_id,
            'reply_to': reply_to or '',
        },
        body=(
            args, kwargs, {
                'callbacks': callbacks,
                'errbacks': errbacks,
                'chain': chain,
                'chord': chord,
            },
        ),
        sent_event={
            'uuid': task_id,
            'root_id': root_id,
            'parent_id': parent_id,
            'name': name,
            'args': argsrepr,
            'kwargs': kwargsrepr,
            'retries': retries,
            'eta': eta,
            'expires': expires,
        } if create_sent_event else None,
    )

4.4 send_task_message in amqp

amqp.send_task_message(P, name, message, **options) 是用来 amqp 发送任务。

该方法主要是组装待发送任务的参数，如connection，queue，exchange，routing_key等，调用 producer 的 publish 发送任务。

基本套路就是：

获得 queue；
获得 delivery_mode；
获得 exchange；
获取重试策略等；
调用 producer 来发送消息;

        def send_task_message(producer, name, message,
                              exchange=None, routing_key=None, queue=None,
                              event_dispatcher=None,
                              retry=None, retry_policy=None,
                              serializer=None, delivery_mode=None,
                              compression=None, declare=None,
                              headers=None, exchange_type=None, **kwargs):

                    # 获得 queue, 获得 delivery_mode, 获得 exchange, 获取重试策略等

            if before_receivers:
                send_before_publish(
                    sender=name, body=body,
                    exchange=exchange, routing_key=routing_key,
                    declare=declare, headers=headers2,
                    properties=properties, retry_policy=retry_policy,
                )

            ret = producer.publish(
                body,
                exchange=exchange,
                routing_key=routing_key,
                serializer=serializer or default_serializer,
                compression=compression or default_compressor,
                retry=retry, retry_policy=_rp,
                delivery_mode=delivery_mode, declare=declare,
                headers=headers2,
                **properties
            )
            if after_receivers:
                send_after_publish(sender=name, body=body, headers=headers2,
                                   exchange=exchange, routing_key=routing_key)

            .....

            if sent_event: # 这里就处理了sent_event
                evd = event_dispatcher or default_evd
                exname = exchange
                if isinstance(exname, Exchange):
                    exname = exname.name
                sent_event.update({
                    'queue': qname,
                    'exchange': exname,
                    'routing_key': routing_key,
                })
                evd.publish('task-sent', sent_event,
                            producer, retry=retry, retry_policy=retry_policy)
            return ret
        return send_task_message

此时堆栈为：

send_task_message, amqp.py:473
send_task, base.py:749
apply_async, task.py:565
<module>, myclient.py:4

此时变量为：

qname = {str} 'celery'
queue = {Queue} <unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>
 ContentDisallowed = {type} <class 'kombu.exceptions.ContentDisallowed'>
 alias = {NoneType} None
 attrs = {tuple: 18} (('name', None), ('exchange', None), ('routing_key', None), ('queue_arguments', None), ('binding_arguments', None), ('consumer_arguments', None), ('durable', <class 'bool'>), ('exclusive', <class 'bool'>), ('auto_delete', <class 'bool'>), ('no_ack', None), ('alias', None), ('bindings', <class 'list'>), ('no_declare', <class 'bool'>), ('expires', <class 'float'>), ('message_ttl', <class 'float'>), ('max_length', <class 'int'>), ('max_length_bytes', <class 'int'>), ('max_priority', <class 'int'>))
 auto_delete = {bool} False
 binding_arguments = {NoneType} None
 bindings = {set: 0} set()
 can_cache_declaration = {bool} True
 channel = {str} 'Traceback (most recent call last):\n  File "
 consumer_arguments = {NoneType} None
 durable = {bool} True
 exchange = {Exchange} Exchange celery(direct)
 exclusive = {bool} False
 expires = {NoneType} None
 is_bound = {bool} False
 max_length = {NoneType} None
 max_length_bytes = {NoneType} None
 max_priority = {NoneType} None
 message_ttl = {NoneType} None
 name = {str} 'celery'
 no_ack = {bool} False
 no_declare = {NoneType} None
 on_declared = {NoneType} None
 queue_arguments = {NoneType} None
 routing_key = {str} 'celery'
  _channel = {NoneType} None
  _is_bound = {bool} False
queues = {Queues: 1} {'celery': <unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>}

此时逻辑如下：

         1  apply_async       +-------------------+
                              |                   |
User  +---------------------> | task: myTest.add  |
                              |                   |
                              +--------+----------+
                                       |
                                       |
                          2 send_task  |
                                       |
                                       v
                                +------+--------+
                                | Celery myTest |
                                |               |
                                +------+--------+
                                       |
                                       |
                  3 send_task_message  |
                                       |
                                       v
                               +-------+---------+
                               |      amqp       |
                               +-------+---------+
                                       |
                                       |
                            4 publish  |
                                       |
                                       v
                                  +----+------+
                                  | producer  |
                                  |           |
                                  +-----------+

4.5 publish in producer

在 produer 之中，调用 channel 来发送信息。

def _publish(self, body, priority, content_type, content_encoding,
             headers, properties, routing_key, mandatory,
             immediate, exchange, declare):
    channel = self.channel
    message = channel.prepare_message(
        body, priority, content_type,
        content_encoding, headers, properties,
    )
    if declare:
        maybe_declare = self.maybe_declare
        [maybe_declare(entity) for entity in declare]

    # handle autogenerated queue names for reply_to
    reply_to = properties.get('reply_to')
    if isinstance(reply_to, Queue):
        properties['reply_to'] = reply_to.name
    return channel.basic_publish( # 发送消息
        message,
        exchange=exchange, routing_key=routing_key,
        mandatory=mandatory, immediate=immediate,
    )

变量为：

body = {str} '[[2, 8], {}, {"callbacks": null, "errbacks": null, "chain": null, "chord": null}]'
compression = {NoneType} None
content_encoding = {str} 'utf-8'
content_type = {str} 'application/json'
declare = {list: 1} [<unbound Queue celery -> <unbound Exchange celery(direct)> -> celery>]
delivery_mode = {int} 2
exchange = {str} ''
exchange_name = {str} ''
expiration = {NoneType} None
headers = {dict: 15} {'lang': 'py', 'task': 'myTest.add', 'id': 'af0e4c14-a618-41b4-9340-1479cb7cde4f', 'shadow': None, 'eta': None, 'expires': None, 'group': None, 'group_index': None, 'retries': 0, 'timelimit': [None, None], 'root_id': 'af0e4c14-a618-41b4-9340-1479cb7cde4f', 'parent_id': None, 'argsrepr': '(2, 8)', 'kwargsrepr': '{}', 'origin': 'gen11468@DESKTOP-0GO3RPO'}
immediate = {bool} False
mandatory = {bool} False
priority = {int} 0
properties = {dict: 3} {'correlation_id': 'af0e4c14-a618-41b4-9340-1479cb7cde4f', 'reply_to': '2c938063-64b8-35f5-ac9f-a1c0915b6f71', 'delivery_mode': 2}
retry = {bool} True
retry_policy = {dict: 4} {'max_retries': 3, 'interval_start': 0, 'interval_max': 1, 'interval_step': 0.2}
routing_key = {str} 'celery'
self = {Producer} <Producer: <promise: 0x1eeb62c44c8>>
serializer = {str} 'json'

此时逻辑为：

         1  apply_async       +-------------------+
                              |                   |
User  +---------------------> | task: myTest.add  |
                              |                   |
                              +--------+----------+
                                       |
                          2 send_task  |
                                       |
                                       v
                                +------+--------+
                                | Celery myTest |
                                |               |
                                +------+--------+
                                       |
                  3 send_task_message  |
                                       |
                                       v
                               +-------+---------+
                               |      amqp       |
                               +-------+---------+
                                       |
                            4 publish  |
                                       |
                                       v
                                  +----+------+
                                  | producer  |
                                  |           |
                                  +----+------+
                                       |
                                       |
                      5 basic_publish  |
                                       v
                                  +----+------+
                                  |  channel  |
                                  |           |
                                  +-----------+

至此一个任务就发送出去，等待着消费者消费掉任务。

4.6 redis 内容

发送之后，task 就被存储在redis的队列之中。在redis 的结果是：

127.0.0.1:6379> keys *
1) "_kombu.binding.reply.testMailbox.pidbox"
2) "_kombu.binding.testMailbox.pidbox"
3) "celery"
4) "_kombu.binding.celeryev"
5) "_kombu.binding.celery"
6) "_kombu.binding.reply.celery.pidbox"
127.0.0.1:6379> lrange celery 0 -1
1) "{\"body\": \"W1syLCA4XSwge30sIHsiY2FsbGJhY2tzIjogbnVsbCwgImVycmJhY2tzIjogbnVsbCwgImNoYWluIjogbnVsbCwgImNob3JkIjogbnVsbH1d\", \"content-encoding\": \"utf-8\", \"content-type\": \"application/json\", \"headers\": {\"lang\": \"py\", \"task\": \"myTest.add\", \"id\": \"243aac4a-361b-4408-9e0c-856e2655b7b5\", \"shadow\": null, \"eta\": null, \"expires\": null, \"group\": null, \"group_index\": null, \"retries\": 0, \"timelimit\": [null, null], \"root_id\": \"243aac4a-361b-4408-9e0c-856e2655b7b5\", \"parent_id\": null, \"argsrepr\": \"(2, 8)\", \"kwargsrepr\": \"{}\", \"origin\": \"gen33652@DESKTOP-0GO3RPO\"}, \"properties\": {\"correlation_id\": \"243aac4a-361b-4408-9e0c-856e2655b7b5\", \"reply_to\": \"b34fcf3d-da9a-3717-a76f-44b6a6362da1\", \"delivery_mode\": 2, \"delivery_info\": {\"exchange\": \"\", \"routing_key\": \"celery\"}, \"priority\": 0, \"body_encoding\": \"base64\", \"delivery_tag\": \"fa1bc9c8-3709-4c02-9543-8d0fe3cf4e6c\"}}"

4.6.1 delivery_tag 作用

可以看到，最终消息中，有一个 delivery_tag 变量，这里要特殊说明下。

可以认为 delivery_tag 是消息在 redis 之中的唯一标示，是 UUID 格式。

具体举例如下:

"delivery_tag": "fa1bc9c8-3709-4c02-9543-8d0fe3cf4e6c"。

后续 QoS 就使用 delivery_tag 来做各种处理，比如 ack, snack。

with self.pipe_or_acquire() as pipe:
    pipe.zadd(self.unacked_index_key, *zadd_args) \
        .hset(self.unacked_key, delivery_tag,
              dumps([message._raw, EX, RK])) \
        .execute()
    super().append(message, delivery_tag)

4.6.2 delivery_tag 何时生成

我们关心的是在发送消息时候，何时生成 delivery_tag。

结果发现是在 Channel 的 _next_delivery_tag 函数中，是在发送消息之前，对消息做了进一步增强。

def _next_delivery_tag(self):
    return uuid()

具体堆栈如下：

_next_delivery_tag, base.py:595
_inplace_augment_message, base.py:614
basic_publish, base.py:599
_publish, messaging.py:200
_ensured, connection.py:525
publish, messaging.py:178
send_task_message, amqp.py:532
send_task, base.py:749
apply_async, task.py:565
<module>, myclient.py:4

至此，客户端发送 task 的流程已经结束，有兴趣的可以看看 [源码解析] 并行分布式任务队列 Celery 之消费动态流程此章从服务端角度讲解收到 Task 如何消费。

0xFF 参考

celery源码分析-Task的初始化与发送任务

Celery 源码解析三：Task 对象的实现

以上是关于[源码分析] 分布式任务队列 Celery 之发送Task & AMQP的主要内容，如果未能解决你的问题，请参考以下文章

[源码分析] 分布式任务队列 Celery 之发送Task & AMQP

[源码解析] 并行分布式框架 Celery 之 Lamport 逻辑时钟 & Mingle

Celery分布式任务

Python 神器 Celery 源码解析：不同启动模式的分析

python之celery队列模块

Django使用celery异步发邮件

[源码分析] 分布式任务队列 Celery 之 发送Task & AMQP

[源码分析] 分布式任务队列 Celery 之 发送Task & AMQP

0x00 摘要

0x01 示例代码

1.1 服务端

1.2 客户端

0x02 系统启动

2.1 产生Celery

2.2 task 装饰器

2.2.1 添加任务

2.2.2 绑定

2.3 小结

0x03 amqp类

3.1 生成

3.2 定义

0x04 发送Task

4.1 apply_async in task

4.2 send_task

4.3 生成消息内容

4.4 send_task_message in amqp

4.5 publish in producer

4.6 redis 内容

4.6.1 delivery_tag 作用

4.6.2 delivery_tag 何时生成

0xFF 参考

[源码分析] 分布式任务队列 Celery 之发送Task & AMQP

[源码分析] 分布式任务队列 Celery 之发送Task & AMQP