8.3 KiB
tags, create time
| tags | create time | |
|---|---|---|
|
2026-05-24 19:52 |
AMQP 协议
概述
AMQP（Advanced Message Queuing Protocol）是应用最广泛的消息队列开放标准之一。本文重点解析 AMQP 0-9-1 版本的核心模型、Exchange 路由机制，并与 1.0 版本进行对比，最后给出 Go 语言的完整收发示例。
正文
AMQP 的版本线
AMQP 并非只有一个版本，理解版本差异是正确使用它的前提：
- AMQP 0-8：早期版本，功能有限，已基本淘汰。
- AMQP 0-9-1：2008 年发布，是 RabbitMQ 实现的版本。定义了完整的 Broker 模型（Exchange/Queue/Binding），是目前使用最广泛的版本。严格来说它不是 OASIS 正式标准，而是社区规范。
- AMQP 0-10：由另一组人维护，与 0-9-1 差异较大，采用者较少（如 Apache Qpid）。
- AMQP 1.0：2012 年成为 OASIS 标准，后纳入 ISO/IEC 19464。这是一个完全重新设计的协议，去掉了 Exchange 等抽象概念，引入 Link/Terminus 模型。Azure Service Bus、ActiveMQ Artemis 等支持此版本。
[!question] 0-9-1 和 1.0 看起来像是两个完全不同的协议。为什么同一个标准组织会允许如此大的断裂？你觉得这是技术进步还是标准碎片化？
AMQP 0-9-1 核心模型
AMQP 0-9-1 的消息流转模型是理解 RabbitMQ 的基础：
graph LR
Producer["Producer"]
Exchange["Exchange"]
QueueA["Queue A"]
QueueB["Queue B"]
ConsumerA["Consumer A"]
ConsumerB["Consumer B"]
Producer -->|"发送消息 + Routing Key"| Exchange
Exchange -->|"Binding Key: order.*"| QueueA
Exchange -->|"Binding Key: log.#"| QueueB
QueueA --> ConsumerA
QueueB --> ConsumerB
核心组件及其职责：
| 组件 | 职责 |
|---|---|
| Connection | TCP 长连接，负责认证和加密 |
| Channel | Connection 内的虚拟连接，多路复用，避免频繁建 TCP |
| Exchange | 消息路由引擎，根据规则将消息分发到 Queue |
| Queue | 消息存储容器，消费者从这里拉取或推送消息 |
| Binding | Exchange 和 Queue 之间的绑定关系，携带路由规则 |
消息投递流程：Producer 将消息发送到 Exchange（不是直接发到 Queue），Exchange 根据自身的类型和 Binding 规则决定消息去往哪些 Queue。这种间接路由模型是 AMQP 最核心的设计思想。
Connection 与 Channel
一个 TCP Connection 上可以复用多个 Channel，这是 AMQP 的性能优化手段：
// 建立一个 Connection
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()
// 在同一 Connection 上打开两个 Channel
ch1, _ := conn.Channel()
ch2, _ := conn.Channel()
// ch1 和 ch2 互不干扰，可以并行操作
为什么需要 Channel？因为每个线程/协程持有独立的 Channel 可以避免锁竞争，而且建立 TCP 连接的开销远大于创建 Channel。
四种 Exchange 类型
Exchange 是消息路由的核心，不同类型的 Exchange 适用于不同的路由策略。
Direct Exchange
精确匹配 Routing Key 和 Binding Key。消息只会到达 Binding Key 完全匹配的 Queue。
graph LR
P["Producer"] -->|"RoutingKey: order.created"| Ex["Direct Exchange"]
Ex -->|"BindKey: order.created"| Q1["Order Queue"]
Ex -->|"BindKey: payment.created"| Q2["Payment Queue"]
适用场景：点对点的消息分发，比如将订单消息路由到订单处理队列。
Fanout Exchange
忽略 Routing Key，将消息广播到所有绑定的 Queue。
graph LR
P["Producer"] -->|"RoutingKey: (忽略)"| Ex["Fanout Exchange"]
Ex --> Q1["Analytics Queue"]
Ex --> Q2["Logging Queue"]
Ex --> Q3["Audit Queue"]
适用场景：日志广播、事件通知，一份消息多个消费者各自处理。
Topic Exchange
基于模式匹配的路由。Binding Key 和 Routing Key 都用 . 分隔单词，支持两个通配符：* 匹配一个单词，# 匹配零或多个单词。
graph LR
P["Producer"] -->|"RoutingKey: order.pay.success"| Ex["Topic Exchange"]
Ex -->|"BindKey: order.*.success"| Q1["Success Queue"]
Ex -->|"BindKey: order.#"| Q2["All Orders Queue"]
Ex -->|"BindKey: log.#"| Q3["Log Queue"]
适用场景：灵活的事件订阅，消费者可以按需订阅感兴趣的消息子集。
Headers Exchange
不依赖 Routing Key，而是根据消息 Headers 中的键值对进行匹配。绑定时指定一组键值对和匹配规则（x-match=all 表示全部匹配，x-match=any 表示任一匹配）。
适用场景：消息属性复杂、需要多维度过滤的场景。实际使用较少，因为 Topic Exchange 在大多数情况下已经够用。
[!question] 四种 Exchange 类型中，Fanout 最简单、Headers 最复杂。但在实际项目中，Direct 和 Topic 的使用频率远高于另外两种。你觉得这是为什么？
Virtual Host：多租户隔离
Virtual Host（vhost）是 AMQP 中的逻辑隔离单元。每个 vhost 拥有独立的 Exchange、Queue 和 Binding 集合，不同 vhost 之间的资源完全隔离。
典型用途：
- 环境隔离：
/dev、/staging、/prod各用一个 vhost - 租户隔离：SaaS 平台为每个租户分配独立 vhost
- 权限控制：可以按 vhost 粒度分配用户权限
// 连接时指定 vhost
conn, _ := amqp.Dial("amqp://user:pass@localhost:5672/production")
AMQP 1.0 vs 0-9-1 的关键差异
AMQP 1.0 并非 0-9-1 的升级版，而是重新设计。核心差异：
| 维度 | 0-9-1 | 1.0 |
|---|---|---|
| 路由模型 | Exchange + Binding + Queue | Link + Terminus（Source/Target） |
| Broker 角色 | Broker 负责路由和存储 | Broker 是可选的，可以点对点 |
| 协议复杂度 | 中等，概念清晰 | 较高，状态机复杂 |
| 生态 | RabbitMQ 生态完善 | Azure、ActiveMQ 支持 |
| 标准化 | 非正式标准 | ISO/IEC 19464 |
1.0 中 Producer 通过 Link 直接连接到 Target（类似 Queue），不再经过 Exchange 路由。这简化了点对点通信，但失去了 0-9-1 灵活的路由能力。
[!question] AMQP 1.0 去掉了 Exchange 概念。如果让你重新设计，你会保留 Exchange 还是去掉它？Exchange 的灵活性和复杂度之间如何权衡？
Go 代码：完整收发示例
使用 amqp091-go 库（RabbitMQ 官方维护的 Go 客户端）演示 Topic Exchange 的消息收发。
发送端：
package main
import (
amqp "github.com/rabbitmq/amqp091-go"
"log"
)
func main() {
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()
ch, _ := conn.Channel()
defer ch.Close()
// 声明 Topic Exchange
ch.ExchangeDeclare("events", "topic", true, false, false, false, nil)
// 声明两个队列
q1, _ := ch.QueueDeclare("order-events", true, false, false, false, nil)
q2, _ := ch.QueueDeclare("log-events", true, false, false, false, nil)
// 绑定：order 相关事件 → order-events 队列
ch.QueueBind(q1.Name, "order.*", "events", false, nil)
// 绑定：所有事件 → log-events 队列
ch.QueueBind(q2.Name, "#", "events", false, nil)
// 发送一条消息，Routing Key 为 order.created
ch.Publish("events", "order.created", false, false,
amqp.Publishing{
ContentType: "application/json",
Body: []byte(`{"order_id": 12345, "amount": 99.9}`),
})
log.Println("消息已发送")
}
接收端：
package main
import (
amqp "github.com/rabbitmq/amqp091-go"
"log"
)
func main() {
conn, _ := amqp.Dial("amqp://guest:guest@localhost:5672/")
defer conn.Close()
ch, _ := conn.Channel()
defer ch.Close()
// 消费 order-events 队列
msgs, _ := ch.Consume("order-events", "", false, false, false, false, nil)
for msg := range msgs {
log.Printf("收到订单消息: %s", msg.Body)
msg.Ack(false) // 手动确认，确保消息不丢失
}
}
代码要点：
ExchangeDeclare声明 Exchange，true表示持久化，Broker 重启后 Exchange 依然存在。QueueBind将 Queue 绑定到 Exchange 并指定 Binding Key，这里用order.*匹配order.created、order.paid等。Consume的autoAck设为false，配合msg.Ack(false)实现手动确认，避免消费者崩溃时消息丢失。