cs-note/hhs/MQ/03-协议与标准/5-AMQP-协议.md

---
tags:
  - MQ
create time: 2026-05-24 19:52
update time: 2026-06-07 20:30
---

# 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, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
if err != nil {
    log.Fatalf("连接失败: %v", err)
}
defer conn.Close()

// 在同一 Connection 上打开两个 Channel
ch1, _ := conn.Channel()
ch2, _ := conn.Channel()
// ch1 和 ch2 互不干扰，可以并行操作
```

为什么需要 Channel？因为每个线程/协程持有独立的 Channel 可以避免锁竞争，而且建立 TCP 连接的开销远大于创建 Channel。

> [!tip] Channel 不是线程安全的
> `amqp091-go` 文档明确指出：Channel **不是线程安全的**。多个 goroutine 不能同时操作同一个 Channel。正确做法是每个 goroutine 持有自己的 Channel，或者加锁。这也是为什么 Connection 上可以开多个 Channel 的设计初衷——一个 goroutine 对应一个 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 的使用频率远高于另外两种。你觉得这是为什么？

> [!tip]- 思考：为什么 Direct 和 Topic 占据"甜蜜点"？
> 核心原因：**Direct + Topic 刚好覆盖了绝大多数路由需求**。
>
> | Exchange | 路由能力 | 问题 |
> |----------|---------|------|
> | **Fanout** | 无差别广播 | 太"粗暴"——要么全发，要么不发，无法做任何过滤 |
> | **Headers** | 多维度 Header 键值匹配 | 太"重"——需要构造 Headers map，匹配语义不直观，调试困难 |
> | **Direct** | 精确匹配 Routing Key | 刚好满足点对点路由，简单直接 |
> | **Topic** | 通配符模式匹配 | 在精确匹配之上加了一层灵活的"订阅"能力 |
>
> 关键洞察：**Topic 是 Direct 的超集**。当 Binding Key 不含通配符时，Topic Exchange 的行为就退化成了 Direct Exchange。所以一个 Topic Exchange 往往就能同时胜任 Direct 的场景，团队只需要掌握一种 Exchange 类型。
>
> **Fanout 为什么用得少？** 场景太窄——"一模一样的消息广播给所有人"的需求在真实业务中不多见。订单事件要广播给积分和通知服务，但不给日志服务——Topic 的 `order.*` 就够了。Fanout 最大的短板是**无法做任何路由过滤**，业务稍微复杂就不够用。
>
> **Headers 为什么用得少？** `x-match: all/any` + 键值对匹配，远不如 `order.*.success` 一目了然；需要检查消息 Headers map，性能和可读性都不如 Routing Key 字符串匹配。**Headers 解决的问题，Topic 用更简单的方式解决了 90%**。
>
> 一句话总结：Direct 处理确定性路由，Topic 处理模式化订阅。两者组合几乎能覆盖所有业务场景，而 Fanout 和 Headers 分别处于"太简单"和"太复杂"的极端，适用面天然就窄。

### 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 的灵活性和复杂度之间如何权衡？

> [!tip]- 思考：Exchange 的去留之争
> **Exchange 的本质是一个发布-订阅解耦层**——把「消息从哪来」和「消息到哪去」彻底分离。
>
> 这个间接层带来的核心能力：
>
> | 能力 | 没有 Exchange 会怎样 |
> |------|---------------------|
> | **一对多路由** | Producer 需要知道所有目标队列，耦合到下游 |
> | **动态订阅** | 新增消费者需要改 Producer 代码或配置 |
> | **路由策略可替换** | 路由逻辑硬编码在 Producer 或 Broker 配置里 |
>
> **AMQP 1.0 为什么去掉 Exchange？** 1.0 的设计哲学是「极简传输层 + 智能端点」：Link/Terminus 模型让 Producer 直连 Target，Broker 成为可选角色，路由逻辑推给应用层。这降低了协议复杂度，但代价是失去了协议级的集中路由能力。
>
> **如果让我重新设计：保留 Exchange 的间接路由思想，但简化实现。** 理由：
>
> 1. **路由是通用需求，不应推给应用层**。当系统有 50 个 Producer、20 个 Queue 时，路由规则散落在各处是运维噩梦。Exchange 把路由集中管理，是正确的关注点分离。
>
> 2. **但 0-9-1 的四种类型是过度设计**。实际项目中 Topic 一种就够了——Fanout 等价于 Binding Key 为 `#` 的 Topic，Direct 等价于不含通配符的 Topic，Headers 极少有人真正需要。
>
> 理想的折中方案：**只保留一种 Pattern-based Router**（类似 Topic），支持通配符匹配，默认不指定 routing key 时等价于 Fanout。
>
> 一句话总结：Exchange 的间接路由思想是正确的——它解决的是真实的架构问题。但 0-9-1 把它做复杂了（4 种类型），1.0 又把它做没了。**协议设计中，「去掉一个概念」和「简化一个概念」是两件完全不同的事。**

### 消息可靠性三件套

在生产环境中，消息丢失是最不能接受的问题。先来看一个具体场景：

> 你在开发电商下单服务。用户点击「下单」后，订单服务发送一条 `order.created` 消息到 MQ，积分服务异步消费这条消息给用户加积分。现在问你三个问题：
>
> 1. 消息发出去了，但 Broker 网络闪断，**Producer 怎么知道消息有没有到？**
> 2. 消息到了 Broker，但 Broker 意外重启，**消息还在不在？**
> 3. 消息推给了 Consumer，Consumer 处理到一半崩了，**积分加了还是没加？**

这三个问题分别对应 AMQP 可靠性保障的三个层次——**生产端确认、Broker 端持久化、消费端确认**。缺少任何一个环节，消息都可能静默丢失。

#### 消息生命周期与丢失窗口

一条消息从 Producer 到 Consumer 的完整生命周期，存在三个关键的「丢失窗口」：

```mermaid
sequenceDiagram
    participant P as Producer
    participant B as Broker
    participant Q as Queue
    participant C as Consumer

    P->>B: 发送消息
    Note over P,B: 窗口 1: 网络丢包或 Broker 崩溃,<br/>Producer 不知道消息是否到达

    B->>Q: 路由并存储
    Note over B,Q: 窗口 2: 消息仅在内存中,<br/>Broker 重启后丢失

    Q->>C: 推送消息
    Note over Q,C: 窗口 3: Consumer 崩溃,<br/>未确认的消息何去何从?

    C->>Q: ACK 确认
    Note over Q,C: 确认后 Broker 才删除消息
```

三件套的分工一目了然：

| 丢失窗口 | 保护机制 | 核心思想 |
|----------|---------|----------|
| 窗口 1：Producer → Broker | **Publisher Confirm** | Broker 告知 Producer「我收到了」 |
| 窗口 2：Broker 存储阶段 | **消息持久化** | 消息落盘，重启不丢 |
| 窗口 3：Broker → Consumer | **手动 ACK** | Consumer 告知 Broker「我处理完了」 |

> [!question]
> 三个窗口中，你觉得哪个窗口的丢失后果最严重？为什么？（提示：想想哪个阶段的丢失完全无感知，连报警都不会触发。）

> [!tip]- 思考：哪个窗口最致命？
> **窗口 2（Broker 存储阶段）的丢失后果最严重。** 它是唯一一个「静默丢失」的窗口——所有参与者都以为消息已经安全了，但实际上它已经消失了。
>
> 逐一分析三个窗口：
>
> | 窗口 | 丢失时谁会察觉？ | 是否会触发报警？ |
> |------|-----------------|----------------|
> | **窗口 1**：Producer → Broker | Producer 会收到 Confirm 超时或 Nack，发送端知道失败了 | ✅ 会——Producer 可以重发、打日志、告警 |
> | **窗口 2**：Broker 存储阶段 | **没有任何一方知道**。Producer 已收到 Confirm，Consumer 还没开始消费 | ❌ 不会——静默丢失 |
> | **窗口 3**：Broker → Consumer | Broker 知道消息没被 ACK，会自动重新入队 | ✅ 会——消息回到队列，不会丢 |
>
> 窗口 2 的致命之处在于**时序上的夹缝**：
>
> 1. Broker 已经收到了消息，返回了 Publisher Confirm → Producer 认为「已送达」✅
> 2. 消息还在内存中，还没刷入 WAL → 磁盘上没有这消息 ❌
> 3. Broker 崩溃重启 → 内存中的消息蒸发，但 Producer 已经认为投递成功，不会重发
> 4. Consumer 从未收到过这条消息，也不知道它曾经存在过
>
> **结果：用户下了单，积分没有加，Producer 日志显示「发送成功」，Broker 重启后日志是全新的——没有任何地方有这条消息的痕迹。**
>
> 这就是为什么 Confirm 的时机应该配置为「消息写入 WAL 之后」，而非仅仅「Broker 收到消息就返回」。把窗口 1 和窗口 2 合并为一个原子操作，才能真正堵住这个静默丢失的漏洞。

#### 生产端保障：Publisher Confirm

Consumer 端有 ACK 机制，那 Producer 怎么确认消息真的到了 Broker？答案是 **Publisher Confirm**——不是 Broker「收到了消息」就确认，而是 Broker **将消息成功路由到所有匹配的 Queue 之后**才返回确认。这是一个容易混淆的细节。

```go
// 开启 Confirm 模式
err := ch.Confirm(false)
if err != nil {
    log.Fatalf("开启 Confirm 失败: %v", err)
}

// 获取确认通知 channel（带缓冲避免阻塞）
confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))

go func() {
    for conf := range confirms {
        if conf.Ack {
            log.Printf("消息 %d 已被 Broker 确认", conf.DeliveryTag)
        } else {
            log.Printf("消息 %d 被 Nack，需要重发", conf.DeliveryTag)
        }
    }
}()
```

代码要点：
- `ch.Confirm(false)` 开启 Confirm 模式。参数 `false` 表示不等待 Broker 的 `basic.ack`（异步模式）。
- `NotifyPublish` 返回一个 channel，Broker 确认消息后会往里写入 `Confirmation`。**这个 channel 必须有足够缓冲**（至少等于并发发送的消息数），否则 Broker 回调时会阻塞。
- `DeliveryTag` 是消息在当前 Channel 内的序号，从 1 开始递增。可以用来匹配发送端的重试逻辑。

> [!question] 为什么 Confirm 比事务更好？
> AMQP 也支持事务（`Tx.Select / Tx.Commit`），但事务是**同步阻塞**的——每条消息必须等上一条的事务提交完成后才能继续发送，吞吐量降低 2-5 倍。Publisher Confirm 是**异步流水线**的——Producer 可以连续发送多条消息，Broker 批量返回 Confirm。这就是为什么生产环境几乎只用 Confirm，不用事务。

> [!tip] 同步 Confirm vs 异步 Confirm
> 上面的代码是异步 Confirm（吞吐量最高）。如果你的场景对吞吐要求不高但想要最简单的写法，也可以用同步 Confirm：
>
> ```go
> // 同步 Confirm：发一条等一条
> ch.Confirm(false)
> confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
>
> ch.PublishWithContext(ctx, "events", "order.created", false, false, msg)
> conf := <-confirms  // 阻塞等待这条消息的确认
> if !conf.Ack {
>     log.Println("Broker 拒绝了这条消息，需要重发")
> }
> ```
>
> 选型建议：吞吐量要求高 → 异步 Confirm；逻辑简单、消息量小 → 同步 Confirm。但**永远不要用事务替代 Confirm**。

#### Broker 端保障：消息持久化

持久化解决的问题很直观：**Broker 重启后消息还在不在？**

但要实现持久化，需要同时满足三个条件，缺一不可：

> [!warning] 持久化的三个条件
> 1. Exchange 声明时 `durable: true` —— Broker 重启后 Exchange 依然存在
> 2. Queue 声明时 `durable: true` —— Broker 重启后 Queue 依然存在
> 3. Message 的 `DeliveryMode` 设置为 `amqp.Persistent`（值为 2）—— 消息体本身要落盘
>
> **只声明了 durable 的 Queue 但消息没有设 Persistent，Broker 重启后消息依然会丢失。** 这是最常见的坑：Queue 配置正确，但发送端忘了设 `DeliveryMode`，导致 Broker 重启后队列还在，消息却没了。

为什么三个条件要同时满足？因为它们是不同层面的东西：

| 条件 | 保护对象 | 如果缺失 |
|------|---------|---------|
| Exchange durable | 路由元数据 | 重启后 Exchange 消失，消息无法路由 |
| Queue durable | 队列元数据 | 重启后 Queue 消失，消息无处存储 |
| Message Persistent | 消息体本身 | 重启后消息从内存清除 |

RabbitMQ 内部的持久化机制是 **WAL（Write-Ahead Log）**：Persistent 消息先写入 WAL 日志，再由后台线程异步刷入 `.rdx` 数据文件。WAL 保证了即使 Broker 崩溃，已写入日志的消息不会丢失。内存中的消息可以被消费者立即读取（不等落盘），兼顾了可靠性和性能。

> [!question] 如果消息在 WAL 写入之前 Broker 就崩了，这条消息会怎样？
> 答案是：**丢了**。这是持久化的物理极限——没有任何系统能保证「零丢失」，只能做到「已确认写入的数据不丢」。这也正是 Publisher Confirm 的意义：**Confirm 的时机可以配置为消息写入 WAL 之后**，而非仅仅是收到消息就返回。这样 Producer 收到 Confirm 时，可以确信消息已经安全落盘。

#### 消费端保障：手动 ACK

消费端是消息生命周期的最后一环，也是最容易出问题的一环。核心问题是：**消费者从 Queue 拿到消息后，如果在处理过程中崩了，这条消息怎么办？**

##### 自动确认 vs 手动确认

AMQP 有两种确认模式，选择不同，消息的命运截然不同：

| 模式 | `autoAck` | 行为 | 风险 |
|------|-----------|------|------|
| **自动确认** | `true` | Broker 发出消息后立即删除 | Consumer 崩溃 → 消息**永久丢失** |
| **手动确认** | `false` | 等 Consumer 发送 Ack 后才删除 | Consumer 崩溃 → 消息自动**重新入队** |

自动确认等于「发了就不管」——Broker 把消息从队列中删掉的那一刻，如果 Consumer 还没处理完就挂了，这条消息就像从来没存在过。**生产环境必须使用手动确认。**

##### Ack / Nack / Reject 三件套

手动确认模式下，Consumer 有三种操作：

| 操作 | 含义 | `multiple` 参数 | `requeue` 参数 |
|------|------|----------------|---------------|
| **Ack** | 处理成功，可以删除 | `true` = 批量确认所有 DeliveryTag ≤ 当前的消息 | 无 |
| **Nack** | 处理失败 | `true` = 批量否定确认 | `true` = 重新入队，`false` = 丢弃或进入死信队列 |
| **Reject** | 处理失败（单条） | 不支持 | `true` = 重新入队，`false` = 丢弃或进入死信队列 |

Nack 和 Reject 的区别：Nack 支持 `multiple=true` 批量操作，Reject 只能操作单条。实际项目中通常用 **Nack**，更灵活。

```go
for msg := range msgs {
    err := processOrder(msg.Body) // 业务逻辑
    if err == nil {
        msg.Ack(false)   // 成功：确认当前这一条
    } else if isTransientError(err) {
        msg.Nack(false, true)   // 临时失败：重新入队，稍后重试
    } else {
        msg.Nack(false, false)  // 永久失败：不重新入队，进入死信队列
    }
}
```

> [!warning] 消息重投递与幂等性
> 当 Consumer 发送 `Nack(false, true)` 或崩溃后消息被自动重新入队时，这条消息会被**再次投递给同一个或另一个 Consumer**。Consumer 端可以通过 `msg.Redelivered` 字段判断是否是重投递。
>
> 这意味着**同一条消息可能被处理两次**。如果你的业务逻辑是「给用户加积分」，重复处理就会重复加积分——这是灾难性的。
>
> **手动 ACK 解决了消息不丢的问题，但引入了消息重复的问题。** 要让整条链路可靠，Consumer 端的业务逻辑必须是**幂等的**——无论处理多少次，结果都一样。常见的幂等方案：
>
> | 方案 | 原理 | 适用场景 |
> |------|------|---------|
> | 数据库唯一约束 | `message_id` 设 UNIQUE，重复插入失败 | 写数据库场景 |
> | Redis SETNX | 用 `message_id` 做 key，成功才执行 | 高并发场景 |
> | 状态机 | 检查当前状态，只有「待处理」才执行 | 订单/支付状态流转 |
>
> **一句话：想要可靠，就必须接受「消息可能重复」，并让业务幂等。这是 AMQP 可靠性模型的基本约束。**

#### 进阶：Prefetch 与公平调度

理解了手动 ACK 之后，你可能会想到一个问题：Consumer 需要处理完一条消息、发送 ACK 之后才能收到下一条——如果 Consumer 处理一条消息需要 1 秒，那它 1 秒内只能处理 1 条，Broker 的 Queue 会持续积压。

更合理的做法是让 Consumer **同时处理多条消息**（比如一个 goroutine 池）。但这就带来了新的风险：如果 Broker 不加限制地推送，慢 Consumer 的缓冲区会堆积大量未处理消息，浪费内存；而快 Consumer 又闲着。

**Prefetch 就是解决这个问题的流控机制。** 它控制 Broker 向单个消费者最多推送多少条**未 ACK 的消息**：

```go
// 每次最多给这个消费者推 10 条未确认消息
err := ch.Qos(
    10,     // prefetchCount：预取数量
    0,      // prefetchSize：预取大小（字节），0 表示不限
    false,  // global：false 表示仅对当前 Channel 生效
)
```

Prefetch 实现了「按能力分配」的公平调度：

```mermaid
graph LR
    Q["Queue<br/>100 条消息"] -->|"推 10 条"| F["快 Consumer<br/>已 ACK 8 条"]
    Q -->|"推 10 条"| S["慢 Consumer<br/>已 ACK 1 条"]
    F -->|"还有 2 条额度"| Q
    S -->|"额度已满, 等待"| Q
```

快 Consumer 处理完 8 条后，Prefetch 额度释放，Broker 立即补充新消息；慢 Consumer 10 条额度一直满着，Broker 不再给它推——**这样就自然实现了负载均衡**。

> [!question] Prefetch 设多少合适？
> 没有标准答案，取决于消息处理耗时。经验公式：`prefetchCount = 单条处理耗时(秒) × 期望吞吐量(条/秒)`。
>
> 例如：单条处理 50ms，期望每秒 200 条 → prefetchCount = 0.05 × 200 = 10。
>
> 设太小（如 1）：Consumer 逐条处理，频繁等待 Broker 推送，吞吐上不去。设太大（如 1000）：回到不设 Prefetch 的老问题。**10-30 是大多数场景的起点，再根据监控微调。**

> [!tip] `global` 参数的陷阱
> `ch.Qos(10, 0, false)` 的第三个参数 `global` 很容易误解：
> - `global=false`（推荐）：Prefetch 限制作用于**单个 Consumer**——每个消费者各自最多 10 条未确认
> - `global=true`：Prefetch 限制作用于**整个 Channel**——该 Channel 上所有消费者共享 10 条额度
>
> 大多数场景应该用 `false`，让每个 Consumer 独立控制。

### Go 代码：完整收发示例

使用 `amqp091-go` 库（RabbitMQ 官方维护的 Go 客户端）演示 Topic Exchange 的消息收发。

> [!warning] API 已更新
> `amqp091-go` v1.7+ 中，`Publish` 已被标记为废弃，应使用 `PublishWithContext` 并传入 `context.Context`，以支持超时和取消。

**发送端**：

```go
package main

import (
	"context"
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
	"time"
)

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("连接失败: %v", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("打开 Channel 失败: %v", err)
	}
	defer ch.Close()

	// 声明 Topic Exchange（durable=true，Broker 重启后保留）
	ch.ExchangeDeclare("events", "topic", true, false, false, false, nil)

	// 声明两个队列（durable=true，持久化队列）
	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)

	// 发送消息，使用 PublishWithContext 替代已废弃的 Publish
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	err = ch.PublishWithContext(ctx, "events", "order.created", false, false,
		amqp.Publishing{
			ContentType:  "application/json",
			DeliveryMode: amqp.Persistent, // 持久化消息，Broker 崩溃后不丢
			Body:         []byte(`{"order_id": 12345, "amount": 99.9}`),
		})
	if err != nil {
		log.Fatalf("发送失败: %v", err)
	}
	log.Println("消息已发送")
}
```

代码要点：
- `ExchangeDeclare` 的第三个参数 `true` 表示持久化，Broker 重启后 Exchange 依然存在。
- `QueueBind` 将 Queue 绑定到 Exchange 并指定 Binding Key，这里用 `order.*` 匹配 `order.created`、`order.paid` 等。
- `PublishWithContext` 传入带超时的 `context.Context`，超时后会返回错误而非永久阻塞。
- `DeliveryMode: amqp.Persistent` 确保消息被写入磁盘，配合 durable 的 Exchange 和 Queue 实现完整的持久化链路。

**接收端**：

```go
package main

import (
	amqp "github.com/rabbitmq/amqp091-go"
	"log"
)

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	if err != nil {
		log.Fatalf("连接失败: %v", err)
	}
	defer conn.Close()

	ch, err := conn.Channel()
	if err != nil {
		log.Fatalf("打开 Channel 失败: %v", err)
	}
	defer ch.Close()

	// 设置 Prefetch：每次最多推 10 条未确认消息
	ch.Qos(10, 0, false)

	// 消费 order-events 队列
	// consumer="" 让 Broker 自动生成唯一 tag
	// autoAck=false，手动确认
	msgs, err := ch.Consume("order-events", "", false, false, false, false, nil)
	if err != nil {
		log.Fatalf("注册消费者失败: %v", err)
	}

	for msg := range msgs {
		log.Printf("收到订单消息: %s", msg.Body)
		// 处理业务逻辑...
		msg.Ack(false) // 手动确认，false 表示仅确认当前这一条
	}
}
```

> [!tip] `Ack(false)` vs `Ack(true)`
> - `msg.Ack(false)` — 只确认当前这一条消息
> - `msg.Ack(true)` — 确认当前消息以及所有 DeliveryTag 比它小的消息（批量确认）
>
> 实际项目中通常用 `false`，逐条确认更安全，避免业务逻辑中途失败时影响已确认的消息。

### 连接恢复策略

网络抖动是生产环境的常态。`amqp091-go` 本身**不提供自动重连**，需要自行实现：

```go
func connectWithRetry(url string) (*amqp.Connection, error) {
    for {
        conn, err := amqp.Dial(url)
        if err == nil {
            log.Println("连接成功")
            return conn, nil
        }
        log.Printf("连接失败: %v，5 秒后重试...", err)
        time.Sleep(5 * time.Second)
    }
}
```

> [!tip] 生产级方案
> 手写重连循环只是入门。生产中更常见的做法是：
> 1. 监听 `conn.NotifyClose()` channel，连接断开时触发重连
> 2. 同时监听 `conn.NotifyBlocked()` 通知 Broker 主动阻塞连接（如内存告警时）
> 3. 在重连后重新声明 Exchange、Queue、Binding 并恢复消费者——因为 AMQP 的声明是**幂等的**（参数一致时不会报错），所以可以放心重复声明

### amqp.Publishing 常用属性

`amqp.Publishing` 结构体有很多字段，生产中常用的几个：

| 属性 | 类型 | 说明 |
|------|------|------|
| `ContentType` | string | 消息体格式，如 `"application/json"` |
| `DeliveryMode` | uint8 | `amqp.Transient`(1) 或 `amqp.Persistent`(2) |
| `MessageId` | string | 消息唯一 ID，用于幂等去重 |
| `CorrelationId` | string | 关联 ID，用于 RPC 场景匹配请求和响应 |
| `ReplyTo` | string | 回复队列名，RPC 模式中告诉服务端把结果发到哪 |
| `Expiration` | string | 消息过期时间（毫秒），如 `"60000"` 表示 1 分钟后过期 |
| `Headers` | map | 自定义键值对，Headers Exchange 根据此字段路由 |

> [!question] RPC 模式
> AMQP 支持天然的 RPC 模式：Client 发消息时设置 `ReplyTo`（回调队列名）和 `CorrelationId`（请求 ID），Server 处理后将结果发到 `ReplyTo` 队列。Client 通过 `CorrelationId` 匹配请求和响应。虽然现在更多用 gRPC 做 RPC，但理解这个模式有助于理解 AMQP 的设计灵活性。

## 关联笔记

- [[4-MQ-消息协议总览]]