239 lines
8.3 KiB
Markdown
239 lines
8.3 KiB
Markdown
---
|
||
tags:
|
||
- MQ
|
||
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 的基础：
|
||
|
||
```mermaid
|
||
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 的性能优化手段：
|
||
|
||
```go
|
||
// 建立一个 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。
|
||
|
||
```mermaid
|
||
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。
|
||
|
||
```mermaid
|
||
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 都用 `.` 分隔单词，支持两个通配符：`*` 匹配一个单词，`#` 匹配零或多个单词。
|
||
|
||
```mermaid
|
||
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 粒度分配用户权限
|
||
|
||
```go
|
||
// 连接时指定 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 的消息收发。
|
||
|
||
**发送端**：
|
||
|
||
```go
|
||
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("消息已发送")
|
||
}
|
||
```
|
||
|
||
**接收端**：
|
||
|
||
```go
|
||
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)` 实现手动确认，避免消费者崩溃时消息丢失。
|
||
|
||
## 关联笔记
|
||
|
||
- [[4-MQ-消息协议总览]]
|