企业官网建设流程全解析

• 任务调度
• 后端
• 消息队列

导读

celery.app.amqp是 Celery 与底层消息代理（Broker）之间的核心桥梁模块，负责任务消息的组装、发布、队列声明与路由决策。本文以官方 API 参考文档 docs/reference/celery.app.amqp.rst 为骨架，结合 celery/app/amqp.py 源码与 t/unit/app/test_amqp.py 测试用例，系统讲解app.amqp的类结构、任务消息协议（v1/v2）、消息发布调用链、队列与路由管理机制，以及相关配置项的默认值与影响。读完本文，你将理解 Celery 一条任务从apply_async到进入 Broker 队列的完整内部路径，并能独立配置队列、交换机、路由与消息协议。

模块定位：应用与 Kombu 之间的消息桥梁

celery.app.amqp位于应用层（celery/app/）内部，是对 kombu 库的封装与整合。从源码第一行注释即可看到其职责：

"""Sending/Receiving Messages (Kombu integration)."""

模块导出的公共符号为__all__ = ('AMQP', 'Queues', 'task_message')，其中：

• AMQP：绑定在app.amqp属性上的核心门面类，向应用暴露队列、交换机、生产者、路由等全部消息能力；
• Queues：一个dict子类，维护“队列名 → kombu.Queue 声明”的映射；
• task_message：一个namedtuple，承载一条待发布任务消息的headers、properties、body与可选的sent_event。

在应用侧，Celery对象通过amqp_cls = 'celery.app.amqp:AMQP'（见 celery/app/base.py）在访问app.amqp时惰性实例化该模块，因此模块中大量使用了cached_property来缓存解析结果。app.amqp也是app.send_task、Task.apply_async、task_routes路由以及 Worker 消费队列声明的最终落点。

AMQP 类的核心属性

官方文档首先定义了AMQP类的若干类级属性，这些属性决定了消息层使用的基础组件：

源码中对应实现（celery/app/amqp.py）：

class AMQP: """App AMQP API: app.amqp.""" Connection = Connection Consumer = Consumer Producer = Producer #: compat alias to Connection BrokerConnection = Connection queues_cls = Queues #: Max size of positional argument representation used for #: logging purposes. argsrepr_maxsize = 1024 #: Max size of keyword argument representation used for logging purposes. kwargsrepr_maxsize = 1024

要点说明：

• Connection/Consumer/Producer均直接引用 Kombu 同名类，并保留了兼容别名BrokerConnection。通过替换这些类属性，可以扩展 Celery 的消息底层行为。
• argsrepr_maxsize/kwargsrepr_maxsize并非截断任务参数本身，而是控制任务在日志、事件中展示时的参数表示长度，避免打印超长参数拖垮日志与监控系统。实际截断逻辑由celery.utils.saferepr.saferepr实现，并且还受task_repr_maxlevels（默认 3）控制嵌套结构的表示层级。
• AMQP.__init__中还注册了任务协议分发表self.task_protocols = {1: self.as_task_v1, 2: self.as_task_v2}，并通过self.app._conf.bind_to(self._handle_conf_update)监听配置更新——当task_routes变化时自动重刷路由表。

任务消息协议：create_task_message 与 v1/v2 两条协议线

create_task_message：按配置选择协议

create_task_message是cached_property，其返回值取决于task_protocol配置（默认 2）：

@cached_property def create_task_message(self): return self.task_protocols[self.app.conf.task_protocol]

也就是说，app.amqp.create_task_message(...)实际上调用的是as_task_v1（协议 1）或as_task_v2（协议 2）之一。两者都返回一个task_messagenamedtuple，但消息的载体结构完全不同。

协议 v2：headers / properties / body 三段式（默认）

as_task_v2（celery/app/amqp.py）是当前默认的消息组装器，它把消息拆成三个部分：

• headers：包含lang、task（任务名）、id（任务 ID）、shadow、eta、expires、group/group_index（组与组内索引）、retries、timelimit（[time_limit, soft_time_limit]）、root_id、parent_id、argsrepr/kwargsrepr（截断后的参数表示，用于日志与事件）、origin（默认取匿名节点名anon_nodename()）、ignore_result、replaced_task_nesting，以及任务打标（stamping）相关的stamped_headers与stamps；
• properties：AMQP 消息属性，至少包含correlation_id（等于任务 ID）与reply_to（回执队列，默认为空字符串）；
• body：一个三元组(args, kwargs, {'callbacks': ..., 'errbacks': ..., 'chain': ..., 'chord': ...})，即任务参数与回调链信息；
• sent_event：仅当create_sent_event=True时非空，用于发布task-sent事件。

v2 协议中countdown与数值型expires会在组装阶段被转换为带时区的eta/expires（通过maybe_make_aware），再序列化为 ISO8601 字符串；而字符串形式的eta/expires（如重试场景）则直接保留。同时 v2 会校验args必须是 list/tuple、kwargs必须是 Mapping，否则抛出TypeError——这一点由测试test_args_must_be_list、test_kwargs_must_be_mapping（t/unit/app/test_amqp.py）覆盖。

协议 v1：扁平 body（旧版兼容）

as_task_v1将所有信息打包进一个扁平 body 字典：task、id、args、kwargs、group/group_index、retries、eta、expires、utc、callbacks、errbacks、timelimit、taskset、chord，headers 为空。v1 是 Celery 4.0 之前的旧协议，仅用于向后兼容；新项目应保持默认的协议 2。

send_task_message：任务消息的发布调用链

send_task_message同样是cached_property，其值来自_create_task_sender()返回的闭包函数（celery/app/amqp.py）。闭包在创建时捕获了全部相关配置与信号引用，以保证每次发布时零开销地取用：

• 默认重试开关task_publish_retry（默认True）；
• 默认重试策略task_publish_retry_policy（默认{'max_retries': 3, 'interval_start': 0, 'interval_max': 1, 'interval_step': 0.2}）；
• 默认投递模式task_default_delivery_mode（默认 2，即持久化）；
• 默认交换机、默认路由键task_default_routing_key、默认序列化器task_serializer（默认'json'）、默认压缩task_compression；
• 三个信号的发送器：before_task_publish、after_task_publish，以及已废弃的task_sent（计划 6.0 移除）。

发布函数send_task_message(producer, name, message, ...)的核心决策逻辑：

1. 确定目标队列/交换机：若未显式传queue与exchange，使用amqp.default_queue；若queue是字符串，则通过queues[queue]解析为kombu.Queue声明。
2. 推导投递模式与交换机类型：优先取队列声明的exchange.delivery_mode与exchange.type，否则回退到配置默认值。
3. 匿名交换机优化：当未指定 exchange 且交换机类型为direct时，直接退化为“匿名交换机 + 队列名作路由键”，即exchange, routing_key = '', qname——这是发送到默认队列时最常见的快速路径；若指定了交换机则使用queue.exchange.name与队列路由键。
4. 合并重试策略：_rp = dict(default_policy, **retry_policy)，自定义策略覆盖默认项。
5. 发出 before 信号：有接收者时才发送before_task_publish。
6. 真正发布：调用producer.publish(...)，参数包含序列化器、压缩、重试策略、投递模式、headers 与 properties 等。
7. 发出 after 信号与事件：有接收者时发送after_task_publish；兼容路径下按协议版本发送废弃的task_sent；若sent_event非空，则通过_event_dispatcher（一个enabled=False的 Dispatcher，配合自定义 producer 使用）发布task-sent事件，事件中补充queue、exchange、routing_key字段。

这条调用链在应用层的入口是app.send_task与Task.apply_async（celery/app/base.py），它们最终都会经由app.amqp.send_task_message完成发布。

队列管理：Queues 类详解

Queues是dict子类，语义为“队列名 ⇒ 队列声明”。文档将其单列为一节，并开启:members:展示全部方法。

构造与默认行为

Queues(queues=None, default_exchange=None, create_missing=True, create_missing_queue_type=None, create_missing_queue_exchange_type=None, autoexchange=None, max_priority=None, default_routing_key=None)

关键参数：

• create_missing（默认True）：遇到未定义队列时自动创建（等价于配置task_create_missing_queues，默认开启）；关闭后访问未知队列会抛KeyError；
• create_missing_queue_type：自动创建队列的类型，仅允许'classic'（默认）或'quorum'，非法值抛ValueError（对应配置task_create_missing_queue_type）；
• create_missing_queue_exchange_type：自动创建队列所用交换机的类型，未设置则用autoexchange（默认kombu.Exchange，对应配置task_create_missing_queue_exchange_type）；
• max_priority：为未显式设置x-max-priority的队列补充该参数（对应配置task_queue_max_priority）。

构造时若传入的是可迭代对象，会先按q.name转成字典，再逐项add()；同时把初始队列集保存为_default_consume_from，作为未使用-Q时的默认消费集合。

队列注册：add 与自动补全

add(queue, **kwargs)接受kombu.Queue实例或队列名字符串；字符串形式走add_compat，其中保留了历史参数binding_key到routing_key的兼容映射，若未指定routing_key则默认取队列名。_add中还会自动补全缺失的交换机（使用default_exchange）与路由键（使用default_routing_key）。__missing__在create_missing=True时调用new_missing(name)动态创建队列——quorum 模式下会设置{'x-queue-type': 'quorum'}。

消费选择：select / deselect / select_add

• select(include)：把消费范围限定为给定队列子集（写入_consume_from），其余队列仅用于路由——对应 Worker 启动时的-Q选项；
• deselect(exclude)：从消费集合中剔除指定队列（也支持按别名剔除）；
• select_add(queue, **kwargs)：显式加入一个“即使有-Q子集也始终消费”的队列。

这些方法在应用层由app.amqp.queues.select(...)暴露，Worker 启动时会调用它处理-Q参数（见 celery/app/base.py）。测试test_select_add、test_deselect、test_deselect_by_alias_removes_selected_queue（t/unit/app/test_amqp.py）覆盖了这些行为。

队列别名与格式化输出

Queues内部维护一个aliases = WeakValueDictionary()，__setitem__时若队列带alias则注册别名，__getitem__会优先查别名——这让同一队列可以拥有可读的短名。format()方法则按QUEUE_FORMAT模板生成路由表的人读日志：

.> queue_name exchange=exchange_name(direct) key=routing_key

Worker 启动时打印的“queues”信息即来自此处。

默认队列、默认交换机与 producer_pool

default_queue 与 default_exchange

@cached_property def default_queue(self): return self.queues[self.app.conf.task_default_queue] @cached_property def default_exchange(self): return Exchange(self.app.conf.task_default_exchange, self.app.conf.task_default_exchange_type)

• default_queue：取配置task_default_queue（默认'celery'）对应的队列声明，是未显式指定队列时任务的默认落点；
• default_exchange：由task_default_exchange与task_default_exchange_type（默认'direct'）构建；task_default_exchange未设置时取task_default_queue的值（见 docs/userguide/configuration.rst）。

producer_pool：生产者连接池

@property def producer_pool(self): if self._producer_pool is None: self._producer_pool = pools.producers[self.app.connection_for_write()] self._producer_pool.limit = self.app.pool.limit return self._producer_pool

producer_pool基于 Kombu 的pools.producers按写连接建池，并复用app.pool.limit作为连接上限；publisher_pool是它的兼容别名。应用层通过app.producer_pool访问（celery/app/base.py），避免每次发布都新建连接。

路由机制：Router、routes 与 flush_routes

文档列出的三个方法Queues()、Router()、flush_routes()共同构成了路由体系：

• AMQP.Queues(queues, ...)：工厂方法，用当前配置（task_create_missing_queues、task_queue_max_priority、task_default_routing_key等）构建Queues实例；若未配置任何队列且存在task_default_queue，会自动用默认交换机、默认路由键构造默认队列，quorum 模式下附带x-queue-type参数；
• Router(queues=None, create_missing=None)：返回celery.app.routes.Router（celery/app/routes.py），它按序执行task_routes中的每条路由规则（字典 →MapRoute，字符串 → 按导入路径解析的路由类），支持'*'通配符与正则，并用lpmerge将命中路由与显式传参合并；expand_destination会把字符串队列名解析为真实kombu.Queue，找不到时抛QueueNotFound；
• flush_routes()：调用_routes.prepare(self.app.conf.task_routes)预编译路由表到_rtable，并在配置更新回调_handle_conf_update中自动重刷。

queues、routes、router均为惰性属性：routes首次访问时flush_routes()，router首次访问时构建Router()。因此只要改task_routes配置，路由表即会随之刷新。

相关配置项速查表

以下配置在 celery/app/defaults.py 中定义，直接影响app.amqp行为：

环境变量方面，这些设置可通过CELERY_DEFAULT_QUEUE、CELERY_DEFAULT_EXCHANGE、CELERY_CREATE_MISSING_QUEUES等前缀形式配置（完整映射见 docs/userguide/configuration.rst）。

实践示例：完整配置一套路由与队列

结合 docs/userguide/routing.rst 与app.amqp的机制，一个典型的多队列路由配置如下：

from kombu import Queue app.conf.task_default_queue = 'default' app.conf.task_default_exchange = 'tasks' app.conf.task_default_exchange_type = 'topic' app.conf.task_default_routing_key = 'task.default' app.conf.task_queues = ( Queue('default', routing_key='task.#'), Queue('feed_tasks', routing_key='feed.#'), ) app.conf.task_routes = { 'feeds.tasks.import_feed': { 'queue': 'feed_tasks', 'routing_key': 'feed.import', }, } app.conf.task_queue_max_priority = 10

该配置在app.amqp.queues中注册两个队列；import_feed任务经Router命中task_routes后路由到feed_tasks队列；未命中路由的任务按默认路由键进入default队列。若希望 Worker 只消费 feed 队列，可启动celery worker -Q feed_tasks，内部即调用app.amqp.queues.select(['feed_tasks'])。

小结

celery.app.amqp是理解 Celery 消息链路的关键模块：AMQP类统一封装了连接（Connection）、消费（Consumer）、生产（Producer）三类基础组件，create_task_message/send_task_message负责按task_protocol组装并发布任务消息（含重试、信号、事件），Queues管理队列声明、自动创建、优先级参数与消费子集选择，Router/routes/flush_routes完成基于task_routes的静态与动态路由，producer_pool则保障高并发下的连接复用。官方参考文档 docs/reference/celery.app.amqp.rst 所列的全部属性与方法，均可在 celery/app/amqp.py 中找到一一对应的实现，配合 t/unit/app/test_amqp.py 中的测试可进一步验证各行为细节。

• 任务调度
• 后端
• 消息队列