vault backup: 2026-08-08 19:01:04
This commit is contained in:
@@ -0,0 +1,179 @@
|
||||
---
|
||||
tags: [mq/rabbitmq, manual-ack, dead-letter-exchange, dlq, retry-pattern, prefetch]
|
||||
create time: 2026-08-08 18:52
|
||||
update time: 2026-08-08 18:52
|
||||
---
|
||||
|
||||
# ACK 确认与死信队列
|
||||
|
||||
## 概述
|
||||
|
||||
Consumer Acknowledgment（ACK）确认机制是 RabbitMQ 消费者可靠性保障的最后防线。Auto ACK 虽然高效但一旦消费者崩溃就会丢消息；Manual ACK 提供了精细的控制能力但增加了复杂度。配合死信队列（DLQ）和重试模式，可以构建出完整的异常消息处理流水线。
|
||||
|
||||
## 核心原理
|
||||
|
||||
### Auto ACK vs Manual ACK
|
||||
|
||||
RabbitMQ 的消费确认模式分为两类：
|
||||
|
||||
| 特性 | Auto ACK | Manual ACK |
|
||||
|------|----------|-----------|
|
||||
| 确认时机 | 消息交付即确认 | 消费者显式调用 basic.ack |
|
||||
| 安全性 | 低（消费者 crash 会丢未处理消息） | 高（处理完才确认） |
|
||||
| 吞吐量 | 略高 | 略低（额外的网络往返） |
|
||||
| 适用场景 | 无状态处理、容忍少量丢失 | 持久化处理、金融类业务 |
|
||||
|
||||
Auto ACK 的隐患在于：AMQP 协议层面，Basic.GetOk / Basic.ConsumeOk 本身就是"投递动作"，而 Auto ACK 把"投递"等同于"处理成功"。如果消费者在收到消息后尚未执行业务逻辑就宕机了，这条消息已经被认为被消费完了。
|
||||
|
||||
Manual ACK 的正确使用方式是在业务逻辑执行完毕后发送 `basic.ack`，如果出现异常则发送 `basic.nack` 并指定 requeue 策略。
|
||||
|
||||
```go
|
||||
// Go 代码示例：Manual ACK 基本模式
|
||||
err := ch.Qos(1, 0, false) // 预取 1 条，公平分发
|
||||
|
||||
msgs, err := ch.Consume("task.queue", "", false, // autoAck=false
|
||||
false, false, false, nil)
|
||||
|
||||
for d := range msgs {
|
||||
result, err := processJob(d.Body)
|
||||
if err != nil {
|
||||
// 业务异常：拒绝并路由到死信
|
||||
ch.Nack(d.DeliveryTag, false, false) // requeue=false
|
||||
continue
|
||||
}
|
||||
ch.Ack(d.DeliveryTag, false) // 处理完成
|
||||
}
|
||||
```
|
||||
|
||||
### unacked 消息的重放机制
|
||||
|
||||
当消费者未发送 ACK 时，消息处于 unacked 状态，存储在 Broker 内存中。此时有三种处置方式：
|
||||
|
||||
1. **basic.ack**：永久标记为已消费，从队列中移除
|
||||
2. **basic.nack(requeue=true)**：放回队列头部，可能被同一消费者再次拉取
|
||||
3. **basic.nack(requeue=false)**：根据队列的 DLX 配置路由到死信队列
|
||||
|
||||
> [!TIP]
|
||||
> 面试高频追问：为什么手动 ACK 会比 Auto ACK 更安全但又可能更慢？答案在于 trade-off——手动 ACK 把控制权交给了消费者，你可以在数据库事务提交后才 ACK，这样即使消费者崩溃，未提交事务对应的消息也不会丢。代价是多了一次 RPC 调用（ack 请求到 Broker）。
|
||||
|
||||
### Dead Letter Exchange（DLX）
|
||||
|
||||
DLX 不是独立的消息队列，而是一种通过队列属性触发的路由行为。当队列配置了 `x-dead-letter-exchange`，满足以下条件的消息会被自动重新发布到指定的 DLX：
|
||||
|
||||
- 被 `basic.nack` 且 `requeue=false`
|
||||
- 消息 TTL 到期（`x-message-ttl` 或单消息 TTL）
|
||||
- 队列达到最大长度（`x-max-length`）
|
||||
|
||||
DLX 本身是一个普通的 Exchange，可以配置自己的 binding key（`x-dead-letter-routing-key`）来控制死信消息的去向。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant P as 生产者
|
||||
participant REX as retry-exchange(Direct)
|
||||
participant RQ as retry-queue<br/>(TTL+DLX)
|
||||
participant DX as dead-letter-exchange
|
||||
participant DQ as dead-letter-queue
|
||||
participant C as 消费者
|
||||
|
||||
P->>REX: publish(order.id=123)
|
||||
REX->>RQ: route to retry-queue<br/>(TTL=30s)
|
||||
RQ->>RQ: 消息等待 TTL 过期
|
||||
Note right of RQ: unacked / 被 nack(requeue=false) / TTL到期
|
||||
RQ->>DX: DLX 捕获消息
|
||||
DX->>DQ: 投递到死信队列
|
||||
DQ->>C: consumer 消费死信
|
||||
```
|
||||
|
||||
### TTL + DLX 实现重试队列模式
|
||||
|
||||
重试是 DLX 最经典的应用场景。通过多层队列接力，实现"尝试 N 次 → 最终死信"的自动降级流程。
|
||||
|
||||
```yaml
|
||||
# 重试队列层级设计
|
||||
retry-layer-1:
|
||||
exchange: retry-exchange
|
||||
queue: retry-queue-1
|
||||
arguments:
|
||||
x-message-ttl: 5000 # 5 秒后重试
|
||||
x-dead-letter-exchange: retry-exchange
|
||||
x-dead-letter-routing-key: retry-layer-2 # 第一次重试仍进 retry-exchange
|
||||
|
||||
retry-layer-2:
|
||||
exchange: retry-exchange
|
||||
queue: retry-queue-2
|
||||
arguments:
|
||||
x-message-ttl: 30000 # 30 秒后重试
|
||||
|
||||
retry-layer-3:
|
||||
exchange: retry-exchange
|
||||
queue: retry-queue-3
|
||||
arguments:
|
||||
x-message-ttl: 120000 # 2 分钟后重试
|
||||
|
||||
dead-letter-queue:
|
||||
exchange: dead-letter-exchange
|
||||
queue: dead-letter-queue
|
||||
arguments:
|
||||
x-dead-letter-exchange: monitoring-exchange # 最终告警
|
||||
```
|
||||
|
||||
这种 N 层队列模式的优势在于每条消息只需要携带一个简单的 TTL header，而不需要维护计数器。当队列配置 `x-dead-letter-routing-key` 指向下一层重试队列所在的 Exchange，消息自然形成递增的时间链。
|
||||
|
||||
> [!NOTE]
|
||||
> 如果不想用多层队列，也可以在消息 body 中嵌入一个 `retryCount` 字段，消费者处理失败时增加计数并重新发送到原 Exchange。这种方式更灵活，但需要应用层维护计数逻辑。
|
||||
|
||||
### Nack + Requeue 策略
|
||||
|
||||
Nack 是 Negative Acknowledgement 的缩写，用于告知 Broker 某条消息不应被继续交给当前消费者。关键的决策树如下：
|
||||
|
||||
- `nack(tag, multiple=false, requeue=true)`：单条重放，回到队列头。适用于临时故障（如下游短暂不可用）。
|
||||
- `nack(tag, multiple=false, requeue=false)`：单条不重放，走 DLX。适用于不可恢复的错误（如数据校验失败）。
|
||||
- `nack(tag, multiple=true, ...)`：一次性拒绝当前通道中所有 unacked 消息。谨慎使用，可能造成大面积消息堆积。
|
||||
|
||||
```go
|
||||
// Go 代码示例：带指数退避的重试 Nack
|
||||
const maxRetries = 3
|
||||
|
||||
func handleMessage(d amqp.Delivery) {
|
||||
retries := d.Headers["x-retry-count"].(int32)
|
||||
if retries >= maxRetries {
|
||||
ch.Nack(d.DeliveryTag, false, false) // 进入死信
|
||||
return
|
||||
}
|
||||
d.Headers["x-retry-count"] = retries + 1
|
||||
|
||||
if err := process(d.Body); err != nil {
|
||||
backoff := math.Pow(2, float64(retries)) * 1000
|
||||
d.Headers["x-delay"] = uint64(backoff)
|
||||
ch.Publish("retry-exchange", "retry", false, false,
|
||||
amqp.Publishing{
|
||||
Body: d.Body,
|
||||
Headers: d.Headers,
|
||||
})
|
||||
ch.Ack(d.DeliveryTag, false)
|
||||
} else {
|
||||
ch.Ack(d.DeliveryTag, false)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
这段代码的关键在于：当处理失败时，先在消息头中递增重试次数，计算指数退避延迟，然后重新发布到重试 Exchange，最后对原消息发送 ACK。这样既避免了 nack 导致的重复拉取，又能让下次投递有时间间隔。
|
||||
|
||||
## 实践场景
|
||||
|
||||
**电商订单支付回调处理**：第三方支付平台发起异步回调时，可能出现网络抖动或服务短暂的负载过高。采用 Manual ACK + 三层重试队列 + DLX 的方案：
|
||||
|
||||
1. 第一层重试间隔 5 秒，适合处理瞬时网络问题
|
||||
2. 第二层重试间隔 30 秒，给下游留出恢复窗口
|
||||
3. 第三层重试间隔 2 分钟，覆盖短时服务重启
|
||||
4. 三次重试均失败后进入死信队列，人工介入或对账修复
|
||||
|
||||
这个模式下，`prefetch_count` 通常设为与消费者并发度相当的值，配合 `no_ack=false` 实现精准控制。
|
||||
|
||||
> [!TIP]
|
||||
> 实践中常见的坑：如果消费者处理耗时过长（比如几分钟），unacked 的消息会持续占用 prefetch 配额，导致其他消费者无法获得新消息。解决方案是将 prefetch 适当放大、或者在处理过程中定期发送 heartbeat ack。
|
||||
|
||||
## 关联笔记
|
||||
- [[Exchange 路由机制]]
|
||||
- [[消息持久化与可靠性投递]]
|
||||
- [[推拉结合消费模式]]
|
||||
@@ -0,0 +1,170 @@
|
||||
---
|
||||
tags: [mq/rabbitmq, exchange-routing, direct-exchange, fanout-exchange, topic-exchange, headers-exchange, delay-exchange]
|
||||
create time: 2026-08-08 18:50
|
||||
update time: 2026-08-08 18:50
|
||||
---
|
||||
|
||||
# Exchange 路由机制
|
||||
|
||||
## 概述
|
||||
|
||||
Exchange（交换机）是 RabbitMQ 消息路由的核心组件，充当生产者与队列之间的中间层。生产者在发送消息时指定 Exchange 名称和 routing key，RabbitMQ 根据 Exchange 类型和路由规则将消息分发到一个或多个 Queue。掌握四种内置 Exchange 类型以及插件扩展的延迟交换能力，是设计可靠消息系统的基础。
|
||||
|
||||
## 核心原理
|
||||
|
||||
### 四种内置 Exchange 类型
|
||||
|
||||
#### Direct Exchange — 精确匹配
|
||||
|
||||
Direct Exchange 使用完全匹配原则：routing key 必须与 Queue 绑定到 Exchange 时的 binding key 完全一致，消息才会被投递。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant P as 生产者
|
||||
participant E["Direct Exchange"]
|
||||
participant Q1["Queue A (key: order.create)"]
|
||||
participant Q2["Queue B (key: order.* )"]
|
||||
participant Q3["Queue C (key: *.pay)"]
|
||||
P->>E: 发送消息<br/>routing key: "order.create"
|
||||
E->>Q1: 匹配成功，投递
|
||||
Note over E,Q1: "order.create" == "order.create"
|
||||
E->>Q2: 匹配失败
|
||||
Note over E,Q2: "order.create" ne "order.*"
|
||||
E->>Q3: 匹配失败
|
||||
Note over E,Q3: "order.create" ne "*.pay"
|
||||
```
|
||||
|
||||
关键特征：一对一或一对多投递（多个 Queue 绑定了相同的 binding key）。这是最常用也最可预测的路由方式。
|
||||
|
||||
#### Fanout Exchange — 广播模式
|
||||
|
||||
Fanout Exchange 忽略 routing key，将消息投递到所有绑定到该 Exchange 的 Queue。每次收到消息就是群发。
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant P as 生产者
|
||||
participant E["Fanout Exchange"]
|
||||
participant Q1["订单队列"]
|
||||
participant Q2["日志队列"]
|
||||
participant Q3["通知队列"]
|
||||
P->>E: 发送消息<br/>routing key: (任意值)
|
||||
E->>Q1: 广播投递
|
||||
E->>Q2: 广播投递
|
||||
E->>Q3: 广播投递
|
||||
Note right of E: 忽略 routing key<br/>全部投递
|
||||
```
|
||||
|
||||
典型场景：事件广播、配置刷新、缓存失效通知。不需要关心消息内容被哪些消费者处理。
|
||||
|
||||
#### Topic Exchange — 通配符匹配
|
||||
|
||||
Topic Exchange 是最灵活的路由模式，支持通配符匹配：
|
||||
- `*`（星号）匹配**恰好一个词**
|
||||
- `#`（井号）匹配**零个或多个词**
|
||||
|
||||
词之间以点号 `.` 分隔。匹配规则示例：
|
||||
|
||||
| routing key | binding key | 是否匹配 |
|
||||
|-------------|------------|---------|
|
||||
| `quick.orange.fox` | `*.orange.*` | 是 |
|
||||
| `quick.orange.fox` | `quick.*.fox` | 是 |
|
||||
| `quick.brown.fox` | `*.orange.*` | 否 |
|
||||
| `quick.orange.male.rabbit` | `#` | 是（匹配所有） |
|
||||
| `lazy.orange.cat` | `lazy.#` | 是 |
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant P as 生产者
|
||||
participant E["Topic Exchange"]
|
||||
participant Q1["Queue: *.create"]
|
||||
participant Q2["Queue: *.pay.*"]
|
||||
participant Q3["Queue: #"]
|
||||
P->>E: routing key: "order.create"
|
||||
E->>Q1: 匹配 "*" -> 投递
|
||||
E->>Q2: "order.pay.*" ne "*.pay.*" -> 不投递
|
||||
E->>Q3: "#" -> 投递
|
||||
P->>E: routing key: "order.pay.success"
|
||||
E->>Q1: "*.create" ne "order.pay.success" -> 不投递
|
||||
E->>Q2: 匹配 "*.pay.*" -> 投递
|
||||
E->>Q3: "#" -> 投递
|
||||
```
|
||||
|
||||
#### Headers Exchange — 基于属性匹配
|
||||
|
||||
Headers Exchange 忽略 routing key，通过检查消息的 headers 属性（key-value 对）来决定路由。可以设置 `match` 参数：`all`（所有 header 都匹配）或 `any`（任一 header 匹配即可）。
|
||||
|
||||
> [!WARNING]
|
||||
> Headers Exchange 性能较差且灵活性不如 Topic Exchange，官方文档明确标注"不推荐在生产中使用"。应优先选择 Topic Exchange。
|
||||
|
||||
### 死信交换机的特殊绑定
|
||||
|
||||
Dead Letter Exchange（DLX）本质上也是一个普通 Exchange，只是通过队列属性间接触发。当队列中的消息满足以下条件之一时，会被重新路由到 DLX：
|
||||
|
||||
- 消息被 nack 且不重新入队（`requeue=false`）
|
||||
- 消息 TTL 过期
|
||||
- 队列长度限制已满
|
||||
|
||||
```yaml
|
||||
# 队列声明时指定 DLX
|
||||
queue:
|
||||
name: "order.processing"
|
||||
arguments:
|
||||
x-dead-letter-exchange: "dead-letter-exchange"
|
||||
x-dead-letter-routing-key: "order.dead"
|
||||
```
|
||||
|
||||
### 延迟交换插件（x-delayed-message）
|
||||
|
||||
RabbitMQ 本身不提供延迟消息功能，需要安装 `rabbitmq_delayed_message_exchange` 插件后使用自定义 Exchange 类型 `x-delayed-message`。它通过消息的 `x-delay` header 控制投递时机。
|
||||
|
||||
```go
|
||||
// Go 代码示例：声明延迟 Exchange
|
||||
// （amqp.go 库简化写法）
|
||||
args := amqp.Table{
|
||||
"x-delayed-type": "direct", // 内部转发使用的 Exchange 类型
|
||||
}
|
||||
ch.ExchangeDeclare(
|
||||
"delayed.exchange", // name
|
||||
"x-delayed-message", // type
|
||||
true, // durable
|
||||
false, // auto-deleted
|
||||
false, // internal
|
||||
false, // no-wait
|
||||
nil, // args
|
||||
args, // 自定义参数
|
||||
)
|
||||
```
|
||||
|
||||
构建消息时设置 `x-delay` header（单位毫秒）：
|
||||
|
||||
```go
|
||||
// Go 代码示例：带延迟头的消息
|
||||
msg := amqp.Publishing{
|
||||
Body: []byte(`{"orderId":"123"}`),
|
||||
Headers: amqp.Table{
|
||||
"x-delay": uint32(60000), // 延迟 60 秒
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
> [!TIP]
|
||||
> 面试常考点：RabbitMQ 原生不支持延迟消息。常见替代方案有定时任务轮询、Redis ZSet 定时弹出、或使用 Kafka 的分区时间排序特性。但在高可靠性场景中，x-delayed-message 插件是最直接可靠的方案。
|
||||
|
||||
## 实践场景
|
||||
|
||||
| 场景 | 推荐 Exchange | 原因 |
|
||||
|------|--------------|-----|
|
||||
| 订单创建通知下游 | Direct | 精确控制消息去向 |
|
||||
| 全局缓存失效广播 | Fanout | 所有服务实例都需要收到 |
|
||||
| 日志分级收集（info/error/warn） | Topic | 按级别前缀灵活路由 |
|
||||
| 支付回调不同金额段分流 | Topic | 按 `amount.linux``amount.high` 等维度拆分 |
|
||||
| 超时未支付取消订单 | x-delayed-message | 精确控制延迟投递时间 |
|
||||
| 死信消息归档 | Direct + DLX 属性 | 标准化异常流程处理 |
|
||||
|
||||
> [!TIP]
|
||||
> 秋招面试技巧：当被问到"如何设计一个支持多种消息类型的路由系统"时，优先考虑 Topic Exchange 的通配符能力，而非为每种类型创建独立的 Direct Exchange。Topic Exchange 天然支持层级化路由，运维上更简洁。
|
||||
|
||||
## 关联笔记
|
||||
- [[消息持久化与可靠性投递]]
|
||||
- [[ACK 确认与死信队列]]
|
||||
- [[推拉结合消费模式]]
|
||||
@@ -0,0 +1,162 @@
|
||||
---
|
||||
tags: [mq/rabbitmq, pull-model, qos, prefetch, backpressure, consumer-balance]
|
||||
create time: 2026-08-08 18:53
|
||||
update time: 2026-08-08 18:53
|
||||
---
|
||||
|
||||
# 推拉结合消费模式
|
||||
|
||||
## 概述
|
||||
|
||||
RabbitMQ 的 Consumer API 名为 `basic.consume`（拉取），但实际运行模型是 Server Push（服务端推送）。这个"名不副实"的设计源于 AMQP 协议的早期约定，也是理解 RabbitMQ 消费模型的关键起点。深入掌握预取策略（QoS）、背压处理和多消费者负载均衡机制，才能在吞吐量和可靠性之间找到最佳平衡点。
|
||||
|
||||
## 核心原理
|
||||
|
||||
### 消费模型的本质
|
||||
|
||||
RabbitMQ 的消费入口确实是拉取操作（`basic.consume` 或 `basic.get`），两者行为截然不同：
|
||||
|
||||
| 模式 | 方法 | 行为 | 特点 |
|
||||
|------|------|------|-----|
|
||||
| 推模式（Push） | `basic.consume` | 建立订阅后，Broker 持续主动投递 | 默认模式，适用于大多数场景 |
|
||||
| 拉模式（Pull） | `basic.get` | 每次调用阻塞获取单条消息 | 适用于管理界面、健康检查等非实时场景 |
|
||||
|
||||
`basic.consume` 的工作流程：
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant C as 消费者
|
||||
participant RMQ as RabbitMQ Broker
|
||||
participant Q as Queue
|
||||
|
||||
C->>RMQ: basic.consume(queue="orders", no_ack=false, prefetch=10)
|
||||
RMQ-->>C: consume-ok(consumer_tag="ct1")
|
||||
Note over C,RMQ: 流控通道已建立
|
||||
loop 消息到达队列
|
||||
RMQ->>C: basic.deliver(tag=1)<br/>queue="orders"
|
||||
C->>C: 处理消息...
|
||||
C->>RMQ: basic.ack(tag=1)
|
||||
C->>RMQ: window refill (available=10)
|
||||
end
|
||||
```
|
||||
|
||||
消费者调用 `basic.consume` 后，RabbitMQ 会在该 Channel 上建立一个流控窗口。每当收到 `basic.deliver` 消息时，窗口减少；消费者发送 `basic.ack` 后窗口恢复。Window 耗尽时，Broker 暂停投递直到窗口回收——这就是**内建背压机制**。
|
||||
|
||||
### Prefetch Count（QoS）设置原理
|
||||
|
||||
`prefetch_count` 定义了单个 Channel 上允许的最大未确认消息数。它是流量控制的调节阀。
|
||||
|
||||
设 prefetch = N 时：
|
||||
|
||||
1. Broker 可以在没有收到 ack 的情况下，最多向该 Channel 发送 N 条消息
|
||||
2. 这 N 条消息累积在未确认（unacked）状态
|
||||
3. 当 ack 一条消息后，unacked 数减一，Broker 再补发一条
|
||||
|
||||
```go
|
||||
// Go 代码示例：Qos 配置
|
||||
// prefetch=1 启用公平分发（Fair Dispatch）
|
||||
ch.Qos(1, // prefetch count
|
||||
0, // prefetch size（字节限制，0表示不限制）
|
||||
false) // global=false 作用于单个 channel
|
||||
```
|
||||
|
||||
为什么 prefetch=1 能实现公平分发？假设消费者 A 处理快、B 处理慢：
|
||||
|
||||
- 如果 prefetch=N（大数值），Broker 可能在短时间内把 N 条消息全部发给 A，而 B 空闲等待
|
||||
- 如果 prefetch=1，A 每处理完一条、发一个 ack，才会收到下一条，自然形成了速度匹配
|
||||
|
||||
> [!WARNING]
|
||||
> `global=true` 是遗留参数，在所有现代客户端中建议设为 `false`。它会对整个连接（而非单个 Channel）生效，在多 Channel 场景下会导致意外行为。
|
||||
|
||||
### 预取策略对吞吐量和内存的影响
|
||||
|
||||
| 预取值 | 吞吐量表现 | 内存占用 | 适用场景 |
|
||||
|--------|-----------|---------|---------|
|
||||
| 1 | 较低（串行节奏） | 最低 | 处理耗时差异大的消费者组 |
|
||||
| 10~50 | 较高 | 中等 | 通用生产环境默认值 |
|
||||
| 100~500 | 最高 | 较高 | 短处理、低延迟需求 |
|
||||
| 0（不设置） | 取决于默认值（通常为 0 即无限） | 可能溢出 | **严禁在生产中使用** |
|
||||
|
||||
prefetch 越大意味着 Broker 可以积压更多消息到消费者内存中。这降低了每次投递的往返开销（减少网络 RTT），但同时增加了消费者的内存压力和崩溃恢复时的损失量。
|
||||
|
||||
### 背压处理机制
|
||||
|
||||
RabbitMQ 的背压在两个层面上工作：
|
||||
|
||||
**Layer 1：Channel 级别的 Window**（prefetch 控制）
|
||||
- 消费者窗口耗尽时，Broker 停止向该 Channel 投递
|
||||
- 消费者恢复正常后自动解封
|
||||
|
||||
**Layer 2：Queue 级别的持久化**
|
||||
- 当消费者整体处理能力低于生产者速度时，消息堆积在 Queue 中
|
||||
- Queue 受 `x-max-length` 约束时可拒绝新消息
|
||||
- 不受限时占用磁盘空间，可能拖垮 Broker
|
||||
|
||||
```
|
||||
生产者速度 > 消费者速度 = 消息在 Queue 中排队等待
|
||||
↓
|
||||
Queue 满 + 有限长 = 新消息被拒
|
||||
↓
|
||||
走 DLX 或触发告警
|
||||
```
|
||||
|
||||
### 多消费者并发与负载均衡
|
||||
|
||||
同一个 Queue 可以有多个消费者（构成 Consumer Group），RabbitMQ 以 Round-Robin 方式将消息均匀分配给各消费者：
|
||||
|
||||
```mermaid
|
||||
graph LR
|
||||
Q["Order Queue"] -->|msg 1| C1["Consumer A"]
|
||||
Q -->|msg 2| C2["Consumer B"]
|
||||
Q -->|msg 3| C1
|
||||
Q -->|msg 4| C2
|
||||
Q -->|msg 5| C1
|
||||
Q -->|msg 6| C2
|
||||
style Q fill:#fff3e0
|
||||
style C1 fill:#e8f5e9
|
||||
style C2 fill:#e8f5e9
|
||||
```
|
||||
|
||||
前提条件是 `prefetch_count` 足够大（否则 prefetch=1 时只有一个消费者活跃，另一个空等）。Round-Robin 的分发方式是平均的但不一定是公平的——处理快的消费者实际上承担了更多工作量，这正是 prefetch=1 要解决的问题。
|
||||
|
||||
### 预取策略调优经验
|
||||
|
||||
高吞吐场景（如日志采集、指标上报）：
|
||||
|
||||
- `prefetch=100~500`，消费者端做好消息批处理
|
||||
- 处理逻辑尽量无状态，避免加锁
|
||||
- Monitor unacked 数量和 Queue depth，发现趋势性增长立即扩容
|
||||
|
||||
低延迟场景（如实时通知、在线游戏）：
|
||||
|
||||
- `prefetch=1~10`，让消费者尽快处理完再拿下一条
|
||||
- 消费者数量尽可能多，分散到不同机器
|
||||
- 监控 p99 延迟而非平均值，个别慢消费者可能拉高尾部
|
||||
|
||||
> [!TIP]
|
||||
> 调优黄金法则：先设 prefetch=1 观察各消费者的处理耗时分布，再根据 p99 耗时最高的那个消费者来估算合适的 batch size，初始设为预估值的 2~3 倍即可。
|
||||
|
||||
## 实践场景
|
||||
|
||||
**高并发商品详情页缓存预热**：每日定时从数据库全量读取商品信息，通过 MQ 推送到缓存集群预热。日终跑批期间数据量大但每条消息处理简单，采用 prefetch=200 配合批量更新 Redis pipeline，在保证不淹没缓存节点的前提下最大化吞吐。
|
||||
|
||||
**即时订单状态推送**：用户下单后需要实时更新前端页面。这里 latency 优先级高于 throughput，设置 prefetch=3，消费者部署 10 个副本分布在不同的 Pod 中，配合 Manual ACK 确保状态变更的可靠性。
|
||||
|
||||
```go
|
||||
// Go 代码示例：根据环境动态调整 Qos
|
||||
func configureQos(env string) int {
|
||||
switch env {
|
||||
case "high-throughput":
|
||||
return 200 // 大批量批处理
|
||||
case "low-latency":
|
||||
return 3 // 追求快速响应
|
||||
default:
|
||||
return 10 // 通用场景
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 扩展阅读
|
||||
- [[Exchange 路由机制]]
|
||||
- [[消息持久化与可靠性投递]]
|
||||
- [[ACK 确认与死信队列]]
|
||||
@@ -0,0 +1,189 @@
|
||||
---
|
||||
tags: [mq/rabbitmq, publisher-confirm, message-persistence, rabbitmq-transaction, return-callback]
|
||||
create time: 2026-08-08 18:51
|
||||
update time: 2026-08-08 18:51
|
||||
---
|
||||
|
||||
# 消息持久化与可靠性投递
|
||||
|
||||
## 概述
|
||||
|
||||
在分布式系统中，消息丢失是比重复消费更严重的问题。RabbitMQ 的可靠性投递涉及三个层面的协同：Exchange 和 Queue 的持久化配置、消息本身的 persistent 标记、以及 Publisher Confirm 机制提供的发送端反馈。理解这些机制的设计动机和取舍，才能在实际业务中做出正确的选型。
|
||||
|
||||
## 核心原理
|
||||
|
||||
### 三阶段持久化
|
||||
|
||||
消息从生产者发出到被消费者消费的整个链路中，任何一个环节未持久化都会导致消息丢失风险。
|
||||
|
||||
```mermaid
|
||||
graph TD
|
||||
A[生产者] -->|"1. Exchange durable"| B["Direct Exchange<br/>(durable=true)"]
|
||||
B -->|"2. Queue durable"| C["Order Queue<br/>(durable=true)"]
|
||||
C -->|"3. Message persistent"| D["磁盘存储"]
|
||||
|
||||
style A fill:#e1f5fe
|
||||
style B fill:#fff3e0
|
||||
style C fill:#e8f5e9
|
||||
style D fill:#fce4ec
|
||||
```
|
||||
|
||||
第一阶段：Exchange 声明时设置 `durable=true`。如果 Exchange 不存在，RabbitMQ 会在启动时自动重建；若未设置 durable，重启后 Exchange 会被删除。
|
||||
|
||||
第二阶段：Queue 声明时设置 `durable=true`。注意：**队列持久化只保证元数据不丢**。实际存储在 queue 中的消息默认为内存页缓存，除非消息带有 persistent 标记才会写入磁盘。
|
||||
|
||||
第三阶段：消息发送时设置 `DeliveryMode=2`（persistent）。持久化消息在抵达 Queue 后会刷盘，而非仅仅留在内存 buffer 中。
|
||||
|
||||
> [!WARNING]
|
||||
> 常见误解：仅设置 Queue durable 并不能保证消息不丢。如果消息 DeliveryMode=1（transient），即使队列本身是持久的，这些瞬态消息在服务器重启时也会被丢弃。
|
||||
|
||||
### Publisher Confirm 机制
|
||||
|
||||
Publisher Confirm 是 RabbitMQ 提供的一种异步确认机制。生产者在开启 confirm 模式后，每条消息都会被分配一个唯一的 `delivery-tag`，Broker 处理完成后会通过回调通知生产者结果。
|
||||
|
||||
两种工作模式：
|
||||
|
||||
| 模式 | 说明 | 适用场景 |
|
||||
|------|------|---------|
|
||||
| 单条确认 | 每发一条消息等一次 ack/nack，串行等待 | 吞吐量要求极低、需严格逐条确认的场景 |
|
||||
| 批量确认 | 一次发送多条，一次性收到确认回调 | **绝大多数场景的首选**，吞吐量接近未开启 confirm 的水平 |
|
||||
|
||||
Confirm 模式的回调有两种触发条件：
|
||||
|
||||
```mermaid
|
||||
sequenceDiagram
|
||||
participant P as 生产者
|
||||
participant RMQ as RabbitMQ Broker
|
||||
participant CC as ConfirmCallback
|
||||
participant RC as ReturnCallback
|
||||
|
||||
P->>RMQ: Publish(msg, mandatory=true)
|
||||
activate RMQ
|
||||
RMQ-->>CC: ack(tag=1001)
|
||||
activate CC
|
||||
Note right of CC: 消息路由到 Queue 成功
|
||||
CC-->>P: 回调确认
|
||||
deactivate CC
|
||||
deactivate RMQ
|
||||
|
||||
Note over P,RMQ: 另一条消息
|
||||
P->>RMQ: Publish(msg2, mandatory=true)
|
||||
activate RMQ
|
||||
alt Exchange存在但无绑定队列
|
||||
RMQ-->>RC: return(msg2)
|
||||
activate RC
|
||||
Note right of RC: 路由失败但非不可恢复
|
||||
RC-->>P: 回调退回
|
||||
deactivate RC
|
||||
else Exchange不存在
|
||||
RMQ-->>RC: return(msg2)
|
||||
activate RC
|
||||
Note right of RC: Exchange 不可达
|
||||
RC-->>P: 回调退回
|
||||
deactivate RC
|
||||
end
|
||||
deactivate RMQ
|
||||
```
|
||||
|
||||
`ConfirmCallback` 告诉生产者消息是否被 Broker 安全接收（路由完成），`ReturnCallback` 处理强制性消息因无法路由而被退回的情况。只有同时开启 `mandatory=true`，ReturnCallback 才会在路由失败时被触发。
|
||||
|
||||
### RabbitMQ Transaction 模式
|
||||
|
||||
RabbitMQ 的 AMQP 协议原生支持事务。开启事务后，发送的消息只有在事务提交后才会真正进入队列。如果事务回滚，期间发送的消息不会进入任何队列。
|
||||
|
||||
事务消息的四项基本原则：
|
||||
|
||||
1. **Channel 必须声明为事务模式**：调用 `channel.txSelect()`
|
||||
2. **消息发送后必须提交事务**：`channel.txCommit()`
|
||||
3. **事务内发送的消息不会立即投递**：直到 txCommit 后才进入队列
|
||||
4. **txCommit 失败则事务回滚**：可通过 `txRollback()` 手动回滚
|
||||
|
||||
```go
|
||||
// Go 代码示例：Transaction 模式（慎用）
|
||||
func sendWithTx(ch *amqp.Channel, body string) error {
|
||||
if err := ch.Tx(); err != nil {
|
||||
return fmt.Errorf("tx select failed: %w", err)
|
||||
}
|
||||
defer func() {
|
||||
_ = ch.TxRollback() // 出错时回滚
|
||||
}()
|
||||
if err := ch.Publish("", "my.queue", false, false,
|
||||
amqp.Publishing{Body: []byte(body)}); err != nil {
|
||||
return err
|
||||
}
|
||||
return ch.TxCommit()
|
||||
}
|
||||
```
|
||||
|
||||
### Confirm vs Transaction：对比分析
|
||||
|
||||
| 维度 | Publisher Confirm | Transaction |
|
||||
|------|------------------|-------------|
|
||||
| 吞吐量 | 高（异步批量） | 低（同步阻塞） |
|
||||
| 实现复杂度 | 简单 | 简单 |
|
||||
| 故障恢复 | 需要重试逻辑 | Broker 自动回滚 |
|
||||
| 资源占用 | 低 | 高（需维护事务日志） |
|
||||
| 语义强度 | 最多一次（至少一次需重试） | 恰好一次（配合幂等设计） |
|
||||
| 适用性 | **生产环境首选** | 几乎不推荐使用 |
|
||||
|
||||
为什么 Confirm 优于 Transaction？
|
||||
|
||||
1. **性能差距巨大**：Transaction 采用同步提交方式，每一条消息的事务提交都要等待 Broker 的 fsync 操作完成，而 Confirm 可以批量异步返回。实测差距可达数倍甚至十倍以上。
|
||||
2. **资源开销更小**：Transaction 需要维护完整的事务日志和回滚状态，在大量并发下容易成为瓶颈。Confirm 模式无需这些额外开销。
|
||||
3. **生态兼容性更好**：Spring AMQP、Go amqp 等主流客户端对 Confirm 的支持远成熟于 Transaction。
|
||||
|
||||
> [!TIP]
|
||||
> 面试回答要点：如果需要"恰好一次"语义，不要指望 RabbitMQ 本身保证——无论是 Confirm 还是 Transaction 都不够。正确做法是：Confirm 作为投递保障 + 业务层幂等键（如 orderId 唯一索引）兜底。这才是工业级方案的思路。
|
||||
|
||||
## 代码示例
|
||||
|
||||
```go
|
||||
// Go 代码示例：Confirm + Return 完整回调链路
|
||||
ch.NotifyPublish(make(chan amqp.Confirmation)) // 注册 Confirm
|
||||
ch.NotifyReturn(make(chan amqp.Return)) // 注册 Return
|
||||
|
||||
confirms := ch.NotifyPublish(nil)
|
||||
for conf := range confirms {
|
||||
if conf.Ack {
|
||||
log.Printf("消息 delivery-tag=%d 已确认", conf.DeliveryTag)
|
||||
} else {
|
||||
// nack 时需要业务侧重试或写入死信表
|
||||
log.Printf("消息 delivery-tag=%d 未被确认", conf.DeliveryTag)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
这段代码展示了 Confirm 的核心结构：通过 `NotifyPublish` 注册一个 Confirmation 通道，每个发出的消息对应一个 Confirmation 对象，其 `Ack` 字段为 true 表示 Broker 已成功处理，false 则需要应用层处理重发或降级逻辑。
|
||||
|
||||
```java
|
||||
// Java 代码示例：Spring AMQP 中的 Confirm 配置
|
||||
@Bean
|
||||
public RabbitTemplate rabbitTemplate(RabbitConnectionFactory factory) {
|
||||
RabbitTemplate template = new RabbitTemplate(factory);
|
||||
template.setMandatory(true); // 开启 ReturnCallback
|
||||
template.setConfirmCallback((correlation, ack, reason) -> {
|
||||
if (!ack) log.warn("消息未确认: {}", reason);
|
||||
});
|
||||
template.setReturnsCallback(returned -> {
|
||||
log.warn("消息被退回: {} - {}", returned.getMessage(), returned.getReason());
|
||||
});
|
||||
return template;
|
||||
}
|
||||
```
|
||||
|
||||
## 实践场景
|
||||
|
||||
**订单支付后的状态推送**：用户支付完成后需要将状态变更推送给物流、库存等多个下游系统。在这个场景下：
|
||||
|
||||
- Exchange 和 Queue 均设置为 `durable=true` 防止服务重启丢数据
|
||||
- 消息设置为 `DeliveryMode=2`（persistent）确保磁盘持久化
|
||||
- 开启 Publisher Confirm，nack 的消息放入本地重试表，定时任务补偿
|
||||
- 配合业务幂等键（支付流水号 + 目标系统标识），确保重复收到的消息不会产生副作用
|
||||
|
||||
> [!WARNING]
|
||||
> 持久化消息会带来明显的性能代价（大约 10 倍吞吐下降）。在对丢消息敏感度低的场景（如埋点日志采集），可以使用 transient 消息获得更高吞吐。关键是评估业务的容错阈值后再决定。
|
||||
|
||||
## 关联笔记
|
||||
- [[Exchange 路由机制]]
|
||||
- [[ACK 确认与死信队列]]
|
||||
- [[推拉结合消费模式]]
|
||||
Reference in New Issue
Block a user