AMQP 1.0

AMQP 1.0 自 RabbitMQ 4.0 起得到原生支持。

版本协商

RabbitMQ 开箱即用地原生支持 AMQP 1.0 和 AMQP 0.9.1，无需任何额外插件。

默认情况下，RabbitMQ 监听端口 5672，接收 AMQP 1.0 和 AMQP 0.9.1 的连接。

在建立 TCP 或 TLS 连接之后、发送任何 AMQP 帧之前，客户端会发送一个协议头，指明其希望使用 AMQP 1.0 还是 AMQP 0.9.1，详见 第 2.2 节 版本协商。

对于 AMQP 1.0 连接，RabbitMQ 要求使用简单认证与安全层（SASL），如 第 5.3 节 SASL 所述。如果客户端不使用 SASL，RabbitMQ 将拒绝该连接，如 图 2.13：协议 ID 拒绝示例 所示。

认证选项

RabbitMQ 的 AMQP 1.0 实现支持多种认证方法

• 用户名和密码对，配合任何支持的认证后端组合（建议与 TLS 配合使用以进行传输层加密）
• JWT 令牌 (OAuth 2)
• 通过 EXTERNAL 认证机制 和 rabbitmq_auth_mechanism_ssl 插件进行的 x.509 证书 认证

协议互操作性

RabbitMQ 支持跨不同协议发布和消费消息，这需要协议转换。

当使用 AMQP 1.0 发布消息时，所有目标队列类型（经典队列、仲裁队列 和 流）都会以其原始的 AMQP 1.0 格式存储消息。如果随后使用 AMQP 1.0 消费该消息，则无需进行协议转换。此外，正如 AMQP 1.0 规范所要求的那样，RabbitMQ 确保了基础消息（bare message）的不可变性。这允许客户端不仅针对消息体，还可以针对 properties 和 application-properties 部分设置消息哈希、校验和及数字签名。

虚拟主机

RabbitMQ 通过虚拟主机支持逻辑多租户。

如果连接应用程序未显式指定虚拟主机，则该连接将使用 rabbitmq.conf 中配置的 default_vhost。

default_vhost = /

AMQP 1.0 客户端可以通过在 open 帧的 hostname 字段值前加上 vhost: 前缀来连接到不同的虚拟主机。

例如，要连接到名为 tenant-1 的虚拟主机，客户端应将 hostname 字段设置为 vhost:tenant-1。

地址

AMQP 1.0 地址 决定了消息发送到哪里或从哪里消费。AMQP 地址引用什么内部对象以及如何解析 AMQP 地址，均未由 AMQP 1.0 规范定义。不同的 AMQP 1.0 代理可以选择以不同方式解释提供的地址。

RabbitMQ 实现了强大且灵活的 AMQP 0.9.1 模型，包括交换器、队列 和 绑定。因此，与 RabbitMQ 通信的 AMQP 客户端将消息发送到交换器，并从队列中消费消息。所以，RabbitMQ 所理解和解析的 AMQP 地址包含交换器名称、队列名称和路由键。

RabbitMQ 4.0 引入了一种新的 RabbitMQ 特定 AMQP 地址格式：v2。旧的 RabbitMQ 3.x 地址格式称为 v1。

重要

AMQP 客户端应使用 v2 地址格式。

地址格式 v1 在 RabbitMQ 4.0 中已被弃用，并将在未来的 RabbitMQ 版本中移除。是否仍支持 v1 格式由弃用功能标志 amqp_address_v1 决定，该标志在 RabbitMQ 4.0 中的弃用阶段为 permitted_by_default。

地址 v2

本节定义了新的 v2 地址格式。

目标地址 v2

可能的 v2 目标地址 格式为：

1. /exchanges/:exchange/:routing-key
2. /exchanges/:exchange
3. /queues/:queue
4. <null>

前三种格式均为 字符串。

第 1 种格式 /exchanges/:exchange/:routing-key 使给定 链路（link） 上的所有消息都被发送到交换器 :exchange，并使用路由键 :routing-key。

第 2 种格式 /exchanges/:exchange 使给定链路上的所有消息都被发送到交换器 :exchange，并使用空路由键 ""。这对于忽略路由键的交换器类型（例如 扇形（fanout） 交换器或 头（headers） 交换器）非常有用。

在前两种格式中设置默认交换器 "" 是不允许的。请改用第 3 种格式。

第 3 种格式 /queues/:queue 使给定链路上的所有消息都被发送到队列 :queue。
该队列必须存在。在内部，此队列目标仍使用默认交换器。因此，用户需要对 amq.default 交换器拥有写入权限。

前 3 种格式要求给定链路上的所有消息的目标地址必须相同。如果需要为同一链路上的不同消息设置不同的交换器、路由键或队列，请使用第 4 种格式。

第 4 种格式是 AMQP null 值。正如 AMQP 扩展 使用 AMQP 匿名终点进行消息路由 中所述，必须设置每条消息的 properties 部分中的 to 字段。允许的 to 地址字符串必须具有相同的格式，即：

1. /exchanges/:exchange/:routing-key
2. /exchanges/:exchange
3. /queues/:queue

其中交换器必须存在。

如果某条消息无法路由（例如因为没有队列绑定到目标交换器），RabbitMQ 将以 released 结果处理该消息。

如果发布应用程序需要将消息发送至：

• 单一目的地：优先使用前三种字符串格式中的一种，而不是第 4 种（null）格式，因为前三种格式性能略好。
• 少量不同的目的地：优先为每个目的地开启一个链路，并使用前三种格式中的一种。
• 大量不同的目的地：优先使用第 4 种（null）格式，并在 to 字段中定义每个目的地。

源地址 v2

唯一有效的 v2 源地址 字符串格式为：

1. /queues/:queue

其中客户端从队列 :queue 消费消息。该队列必须存在。

百分号编码（Percent-encoding）

地址格式 v2 要求交换器名称、路由键和队列名称根据 RFC 3986 进行百分号编码。

例如，客户端若想发送至交换器 amq.direct 且路由键为 my-routing_key/123，则必须使用目标地址 /exchanges/amq.direct/my-routing_key%2F123。

请注意，地址格式 v2 中的百分号编码必须应用于所有需要 address 的 AMQP 字段：

• target 中的 address 字段
• source 中的 address 字段
• 消息 properties 中的 to 字段
• 消息 properties 中的 reply-to 字段

地址 v1

警告

本节列出了已弃用的 v1 地址字符串格式。

目标地址 v1

1. /exchange/:exchange/:routing-key
2. /exchange/:exchange
3. /topic/:routing-key
4. /amq/queue/:queue
5. /queue/:queue
6. :queue
7. /queue

第 1 种格式 /exchange/:exchange/:routing-key 使给定链路上的所有消息都被发送到交换器 :exchange，路由键为 :routing-key。对应的 v2 格式是 /exchanges/:exchange/:routing-key。

第 2 种格式 /exchange/:exchange 使给定链路上的所有消息都被发送到交换器 :exchange，路由键可选地在消息 properties 部分的 subject 字段中提供。在 v2 中，为每条消息定义不同的路由键需要将目标地址设置为 AMQP null 值，并将消息的 to 字段设置为 /exchanges/:exchange/:routing-key。

第 3 种格式 /topic/:routing-key 使给定链路上的所有消息都被发送到 RabbitMQ 的默认主题交换器 amq.topic，主题为 routing-key。在 v2 中，使用 /exchanges/amq.topic/:routing-key。

第 4 种格式 /amq/queue/:queue 使给定链路上的所有消息被发送到队列 :queue（更准确地说，在内部是发送到带有路由键 :queue 的默认交换器）。队列 :queue 必须存在。在 v2 中，使用 /queues/:queue。

第 5 种格式 /queue/:queue 与第 4 种格式语义相似。然而，RabbitMQ 会自动声明队列 :queue，即如果该队列不存在则创建一个。该队列永远不会被 RabbitMQ 自动删除。在 v2 中，使用 /queues/:queue。RabbitMQ 4.0 允许 AMQP 客户端创建包括队列在内的 RabbitMQ 拓扑结构，包括用户自定义的队列类型、属性和参数。因此，RabbitMQ 本身无需再为给定的队列目标地址格式自动声明特定的队列。

第 6 种格式 :queue 是第 5 种格式的冗余。

第 7 种格式使消息被发送到消息 subject 字段中提供的队列。在 v2 中，要将消息发送到不同的队列，请将目标地址设置为 AMQP null 值，并将消息的 to 字段设置为 /queues/:queue。

源地址 v1

1. /exchange/:exchange/:binding-key
2. /topic/:binding-key
3. /amq/queue/:queue
4. /queue/:queue
5. :queue

第 1 种格式 /exchange/:exchange/:binding-key 使 RabbitMQ 声明一个队列，并将该队列绑定到交换器 :exchange，绑定键为 :binding-key。随后从该队列中消费消息。

第 2 种格式 /topic/:binding-key 使 RabbitMQ 声明一个队列，并将该队列绑定到默认主题交换器 amq.topic，主题过滤器为 :binding-key。随后从该队列中消费消息。

第 3 种格式 /amq/queue/:queue 使 RabbitMQ 从队列 :queue 中消费。队列 :queue 必须存在。

第 4 种格式 /queue/:queue 使 RabbitMQ 声明一个队列 :queue 并从中消费。

第 5 种格式 :queue 是第 4 种格式的冗余。

如前所述，RabbitMQ 4.0 允许 AMQP 客户端创建包括队列在内的 RabbitMQ 拓扑结构，包括用户自定义的队列类型、属性和参数。因此，RabbitMQ 本身无需再为给定的队列源地址格式自动声明特定的队列。在 v2 中，客户端应首先声明自己的队列和绑定，然后使用源地址 /queues/:queue 进行连接，从而从该队列中消费。

消息注解

当消息投递给消费者时，RabbitMQ 至少会设置以下两个 消息注解：

• x-exchange：设置为消息最初发布到的交换器。
• x-routing-key：设置为消息最初发布时所用的路由键。

这些消息注解根据消息最初发送到 RabbitMQ 的方式，以不同方式推导出来。例如，它们可以推导自：

• AMQP 1.0 发布者所挂载的 target 的 address 字段。
• AMQP 1.0 消息 properties 部分中的 to 字段。
• 如果消息最初通过 AMQP 0.9.1 发布，则为 AMQP 0.9.1 basic.publish 帧中的 exchange 和 routing_key 字段。
• 如果消息最初通过 MQTT 发布，则为 MQTT PUBLISH 数据包的 主题名称。
• 如果消息最初通过 RabbitMQ 流协议 发布，则为流的名称。

然而，AMQP 1.0 客户端不应向 RabbitMQ 发布带有 x-exchange 或 x-routing-key 消息注解的消息。RabbitMQ 不会解释它们。相反，如果 AMQP 1.0 客户端想要将消息重新发布到原始交换器并使用原始路由键，则应相应地设置地址。

结果（Outcomes）

结果（Outcome） 表示接收方处理投递（消息）的结果。

下表描述了当客户端作为发送者/发布者/生产者而 RabbitMQ 作为接收者时的结果：

AMQP 1.0 结果	等效 AMQP 0.9.1 帧	描述
Accepted	basic.ack	消息被路由到的所有队列均已接受该消息。例如，对于 仲裁队列，这意味着多数仲裁队列副本已将消息写入磁盘。因此，发布者可以遗忘/删除该消息。
Rejected	basic.nack	至少有一个队列拒绝了该消息，原因可能是超出了队列长度限制（当 溢出行为 设置为 reject-publish 时），或者因为目标 经典队列 不可用。error 字段的 info 映射中包含键 queue（值为队列名称）和 reason（值为 maxlen 或 unavailable）。 RabbitMQ 还会按照 使用 AMQP 匿名终点进行消息路由 中的规定拒绝消息，例如如果消息 properties 部分的 to 字段包含无效地址或定义了不存在的交换器。
Released	basic.return (随后是 basic.ack 或 basic.nack)	RabbitMQ 无法将消息路由到任何队列。这表明拓扑配置有误，例如没有匹配的队列绑定到目标交换器。
Modified		目前，RabbitMQ 不会以 Modified 结果来结算消息。

下表描述了当客户端作为接收者/消费者而 RabbitMQ 作为发送者时的结果：

AMQP 1.0 结果	等效 AMQP 0.9.1 帧	描述
Accepted	basic.ack	消费者已成功处理该消息。因此 RabbitMQ 可以删除该消息。
Rejected	basic.nack 或 basic.reject (requeue=false)	消费者表明该消息无效且无法处理。RabbitMQ 对消息执行死信处理（如果未配置死信处理，则丢弃该消息）。
Released	basic.nack 或 basic.reject (requeue=true)	消费者未处理该消息。RabbitMQ 对消息执行重新入队。该消息将投递给相同或不同的消费者。
Modified		消费者未处理该消息，但修改了 消息注解。 如果 undeliverable-here=true，RabbitMQ 对消息执行死信处理（如果未配置死信处理，则丢弃该消息）。 如果 undeliverable-here=false，RabbitMQ 对消息执行重新入队。 更多信息请见下文。

AMQP 1.0 vs. AMQP 0.9.1

顾名思义，AMQP 1.0 是更现代的协议。它是 ISO/IEC 19464 和 OASIS 标准，而 AMQP 0.9.1 不是官方标准。关于协议的详细比较，请参考我们的 AMQP 1.0 博客文章。

选择合适的协议取决于多个因素，包括：

• 功能需求：您是否需要 AMQP 1.0 或 AMQP 0.9.1 的特定功能。
• 互操作性：如果与其他消息代理的互操作性很重要，请注意支持 AMQP 1.0 的代理多于 AMQP 0.9.1。
• 客户端库可用性：您的编程语言是否有受支持的客户端库。

AMQP 1.0 功能

本节列出了 RabbitMQ 仅在 AMQP 1.0 中支持而 AMQP 0.9.1 中不可用的功能：

• 细粒度流控：详见博客文章 AMQP 1.0 流控的十大优势。
  • 消费客户端应用程序可以动态调整并确定从特定源队列接收消息的优先级。
  • 在单个 AMQP 连接上同时进行发布和消费的安全且高效的使用方式。
  • 当一个目标队列负载过重时，发布者可以继续高速发送至其他目标队列，消费者也可以在同一 AMQP 连接上从其他源队列高速接收消息。
  • 消费者可以停止或暂停，稍后再恢复。
  • 在保持消息顺序的同时，实现从一个单一活动消费者到下一个消费者的平滑移交。
  • 源队列可以有效地向消费者告知大约可用的消息数量。
• AMQP 过滤表达式：RabbitMQ 在通过 AMQP 1.0 消费流（stream）时实现了 AMQP 过滤表达式。
  • 服务端对复杂 SQL 表达式的评估。
  • 将布隆过滤器与 AMQP 过滤表达式相结合时，RabbitMQ 允许进行高效的分块级过滤，随后再针对复杂业务逻辑进行精确的消息级过滤——全部在服务端完成。
  • 仅调度客户端真正感兴趣的消息，减少了 RabbitMQ 与客户端之间的网络流量。
  • 允许多个并发客户端在保持消息顺序的同时各自只消费一部分消息。
• 队列局部性（Queue Locality）：RabbitMQ 可以向客户端提供最新的队列拓扑和领导者信息。
  • 例如，RabbitMQ AMQP 1.0 Java 客户端 可以利用这些信息，尝试从托管队列副本的 RabbitMQ 节点“本地”消费，并尝试发布到托管队列领导者的节点“本地”。
  • 这可以减少集群内流量，降低延迟并提高吞吐量。
• WebSocket：VMware Tanzu RabbitMQ 支持 基于 WebSocket 的 AMQP 1.0，允许在浏览器中运行的应用程序使用 AMQP 1.0 与 RabbitMQ 通信。
• 修改结果（Modified Outcome）：允许仲裁队列的消费者在重新入队或对消息进行死信处理时添加和修改 消息注解。
• 发送者结算模式 mixed：允许发布者按每条消息决定是否从代理接收 确认（confirmations）。
• 详细的拒绝信息：当 RabbitMQ 拒绝消息时，发布者会在 Rejected 结果 中收到队列名称和拒绝原因，从而识别出具体是哪个队列拒绝了消息以及原因。当多个队列绑定到同一个目标交换器时，此功能特别有用。
• 链路状态属性：RabbitMQ 通过 AMQP 1.0 flow 帧中的属性向消费者传达链路状态信息。例如，当从仲裁队列消费时，若启用了单一活动消费者，消费者会收到通知，以了解自己是活动消费者还是非活动（等待）消费者。
• 定义明确的 类型系统
• 定义更完善的 消息头
• 增强的消息完整性：由于基础消息是不可变的，客户端可以不仅针对消息体，还可以针对 properties 和 application-properties 部分设置消息哈希、校验和及数字签名。
• 流消息保真度：当从 流 存储或检索消息时，不会丢失头信息的保真度，因为流以 AMQP 1.0 编码格式存储消息。

AMQP 0.9.1 功能

本节列出了 RabbitMQ 仅在 AMQP 0.9.1 中支持而当前 AMQP 1.0 中不可用的功能：

• AMQP 0.9.1 通道拦截器：诸如 Sharding 插件 等拦截并修改帧的插件，目前仅支持 AMQP 0.9.1。
• 管理界面中的消息速率：AMQP 0.9.1 连接在管理界面中会显示交换器和队列的消息速率。这些速率在 AMQP 1.0 连接中不可用。
• 事务：AMQP 0.9.1 提供了有限的支持，而 AMQP 1.0 目前不支持事务（如局限性中所列）。

客户端

任何 AMQP 1.0 客户端都应能够与 RabbitMQ 通信。Broadcom 的 RabbitMQ 团队已开发了两个专为 RabbitMQ 设计的 AMQP 1.0 客户端库。

• RabbitMQ AMQP 1.0 **Java** 客户端
• RabbitMQ AMQP 1.0 **.NET** 客户端

更多信息请参见 AMQP 客户端库页面。

目前，AMQP 0.9.1 客户端生态系统更为广泛，Broadcom 的 RabbitMQ 团队支持更多数量的 AMQP 0.9.1 客户端库。

链路状态属性

RabbitMQ 使用 AMQP 1.0 flow 帧的 properties 字段向消费者传达链路状态信息。

目前，支持以下链路状态属性：

属性	类型	描述
rabbitmq:active	布尔值	该消费者是否处于活动状态并将接收消息。true 意味着消费者是活动的。false 意味着消费者处于非活动（等待）状态，因为启用了 单一活动消费者 (SAC)，并且另一个消费者当前处于活动状态。

rabbitmq:active 属性在 flow 帧中发送给 仲裁队列 的每个消费者：

• 在消费者首次授予信用额度后立即发送，指明初始活动状态。对于未启用 SAC 的仲裁队列，值始终为 true，因为每个消费者都是活动的。
• 每当消费者的活动状态发生变化时，例如当优先级更高的消费者连接到已启用 SAC 的仲裁队列，或活动消费者断开连接时。

限制

RabbitMQ 不支持以下 AMQP 1.0 功能：

• “挂起（Suspending）”或“恢复（resuming）”链路，包括：
  • 图 2.8：链路恢复
  • “恰好一次（exactly once）”投递
  • 恢复投递
  • 终点过期策略（Terminus Expiry Policy）
• transfer 帧中的 aborted 字段
• 事务
• 用于 TLS 安全层的协议头（图 5.1），包括协议 ID 为 2 的情况。相反，RabbitMQ 运行纯 TLS 服务器，因此实现了 第 5.2.1 节。

修改结果（Modified Outcome）

通过 Modified 结果修改消息注解的功能在 仲裁队列 中受支持，但在 经典队列 中不受支持。鉴于流是一个不可变的日志，修改流中的消息没有任何意义。

如果 undeliverable-here 字段为：

• true：经典队列和仲裁队列将对消息执行死信处理。如果未配置死信处理，消息将被丢弃。
• false：经典队列和仲裁队列将对消息执行重新入队。

AMQP 1.0 修改结果 博客文章描述了相关用例。

警告

undeliverable-here 的行为在未来的 RabbitMQ 版本中可能会发生变化。

例如，如果 undeliverable-here = true，将来队列可能会在重新入队消息的同时确保消息不会被重新投递给进行修改的链路端点，而不是对消息进行死信处理。