vault backup: 2026-06-03 10:30:42

hzh/Gen2D/04-RabbitMQ集成.md

@@ -1,17 +1,28 @@
-# 04 - RabbitMQ 集成
+---
+tags: [rabbitmq, amqp, message-queue, persistence, ack-nack, retry-pattern, go]
+create time: 2026-06-03 10:15
+---
+
+# 04. RabbitMQ 集成
+
+## 概述
+
+基于 AMQP 协议的持久化消息队列实现，支持手动 ACK/NACK、失败重试和死信丢弃，保障消息不丢失。
 
 > **一句话概括**：基于 AMQP 的持久化消息队列，支持手动 ACK、失败重试和死信丢弃。
 
-## 消息流
+## 正文
+
+### 消息流
 
 ```mermaid
 graph TB
     subgraph Producer["生产者"]
-        HANDLER["Handler.Generate()"]
+        HANDLER["Handler.Generate"]
     end
 
     subgraph RabbitMQ["RabbitMQ"]
-        EXCHANGE["Default Exchange<br/>(Direct)"]
+        EXCHANGE["Default ExchangeDirect"]
         QUEUE["gen2d:tasks<br/>durable=true"]
     end
 
@@ -19,15 +30,15 @@ graph TB
         CONSUME["channel.Consume<br/>autoAck=false"]
     end
 
-    subgraph Decision["ACK/NACK 决策树"]
-        SUCCESS{"handler 成功?"}
-        RETRY{"retry < maxRetry?"}
-        ACK_OK["ACK<br/>确认消费"]
+    subgraph Decision["ACK/NACK决策树"]
+        SUCCESS{"handler成功?"}
+        RETRY{"retry lt maxRetry?"}
+        ACK_OK["ACK确认消费"]
         NACK["NACK + requeue<br/>重新入队"]
-        ACK_DISCARD["ACK (discard)<br/>丢弃死信"]
+        ACK_DISCARD["ACK discard<br/>丢弃死信"]
     end
 
-    HANDLER -->|"Publish<br/>Persistent"| EXCHANGE
+    HANDLER -->|"Publish Persistent"| EXCHANGE
     EXCHANGE --> QUEUE
     QUEUE --> CONSUME
     CONSUME --> SUCCESS
@@ -38,24 +49,16 @@ graph TB
     NACK -.->|"重新投递"| QUEUE
 ```
 
-## 连接流程
+### 连接流程
 
 RabbitMQQueue 在初始化时完成连接、声明队列、设置 QoS：
 
-```
-amqp.Dial(cfg.URL)
-    │
-    ▼
-conn.Channel()
-    │
-    ▼
-ch.QueueDeclare(name, durable=true, autoDelete=false, exclusive=false)
-    │
-    ▼
-ch.Qos(prefetch=1, prefetchSize=0, global=false)
-    │
-    ▼
-RabbitMQQueue{conn, channel, queue, maxRetry, prefetch}
+```mermaid
+graph TB
+    DIAL["amqp.Dialcfg.URL"] --> CHANNEL["conn.Channel"]
+    CHANNEL --> DECLARE["ch.QueueDeclare name durable=true"]
+    DECLARE --> QOS["ch.Qosprefetch=1"]
+    QOS --> RESULT["RabbitMQQueue实例"]
 ```
 
 **参数说明**：
@@ -67,9 +70,11 @@ RabbitMQQueue{conn, channel, queue, maxRetry, prefetch}
 | `Prefetch` | `1` | 每次预取消息数，1 保证公平调度 |
 | `MaxRetry` | `3` | 失败最大重试次数 |
 
-> :bulb: **Prefetch=1 的含义**：每个 Consumer 同时只处理 1 条消息，处理完（ACK）后才接收下一条。这避免了消息堆积在 Consumer 端，配合协程池的并发控制实现精确的任务调度。
+> [!tip] Prefetch=1 的含义
+>
+> 每个 Consumer 同时只处理 1 条消息，处理完ACK后才接收下一条。这避免了消息堆积在 Consumer 端，配合协程池的并发控制实现精确的任务调度。
 
-## 消息发布 (Submit)
+### 消息发布 (Submit)
 
 ```go
 func (q *RabbitMQQueue) Submit(ctx context.Context, msg TaskMessage) error {
@@ -100,7 +105,7 @@ func (q *RabbitMQQueue) Submit(ctx context.Context, msg TaskMessage) error {
 | `ContentType` | `application/json` | JSON 序列化 |
 | `x-retry-count` | `int` (header) | 当前重试次数，供消费端判断 |
 
-## 消息消费 (Consume)
+### 消息消费 (Consume)
 
 ```go
 func (q *RabbitMQQueue) Consume(ctx context.Context, handler func(TaskMessage) error) error {
@@ -121,25 +126,25 @@ func (q *RabbitMQQueue) Consume(ctx context.Context, handler func(TaskMessage) e
 
 手动 ACK 给予消费者完全的控制权——只有当消息被成功处理后才确认，否则可以选择重试或丢弃。
 
-## ACK/NACK 决策树
+### ACK/NACK 决策树
 
 ```mermaid
 graph TB
-    MSG["收到消息"] --> PARSE{"JSON 解析成功?"}
-    PARSE -->|否| ACK_DISCARD1["ACK (discard)<br/>格式错误无法恢复"]
-    PARSE -->|是| HANDLER{"handler(msg)<br/>执行成功?"}
-    HANDLER -->|是| ACK_OK["ACK<br/>确认消费"]
-    HANDLER -->|否| RETRY_CHECK{"msg.RetryCount<br/>< maxRetry?"}
-    RETRY_CHECK -->|是| NACK["NACK(requeue=true)<br/>重新入队等待重试"]
-    RETRY_CHECK -->|否| ACK_DISCARD2["ACK (discard)<br/>超过最大重试，记录死信日志"]
+    MSG["收到消息"] --> PARSE{"JSON解析成功?"}
+    PARSE -->|否| ACK_DISCARD1["ACK discard<br/>格式错误无法恢复"]
+    PARSE -->|是| HANDLER{"handlermsg执行成功?"}
+    HANDLER -->|是| ACK_OK["ACK确认消费"]
+    HANDLER -->|否| RETRY_CHECK{"msg.RetryCount lt maxRetry?"}
+    RETRY_CHECK -->|是| NACK["NACKrequeuetrue<br/>重新入队等待重试"]
+    RETRY_CHECK -->|否| ACK_DISCARD2["ACK discard<br/>超过最大重试记录死信日志"]
 ```
 
 | 场景 | 操作 | 说明 |
 |------|------|------|
-| handler 成功 | `d.Ack(false)` | 确认消费，消息从队列移除 |
-| handler 失败 + retry < max | `d.Nack(false, true)` | 拒绝并重新入队，retry count 递增 |
-| handler 失败 + retry >= max | `d.Ack(false)` + 日志 | 超过最大重试，丢弃（可扩展为死信队列） |
-| JSON 解析失败 | `d.Ack(false)` | 格式错误无法恢复，直接丢弃 |
+| handler 成功 | `d.Ackfalse` | 确认消费，消息从队列移除 |
+| handler 失败 + retry < max | `d.Nackfalse, true` | 拒绝并重新入队，retry count 递增 |
+| handler 失败 + retry >= max | `d.Ackfalse` + 日志 | 超过最大重试，丢弃可扩展为死信队列 |
+| JSON 解析失败 | `d.Ackfalse` | 格式错误无法恢复，直接丢弃 |
 
 **重试计数传递**：
 
@@ -155,9 +160,11 @@ if retry, ok := d.Headers["x-retry-count"].(int32); ok {
 }
 ```
 
-> :warning: **NACK requeue 的行为**：`Nack(false, true)` 会将消息重新放回队列头部。如果消费者立即再次消费，可能导致"毒消息"反复重试。Gen2D 通过 `maxRetry=3` 限制重试次数，并在超过后 ACK 丢弃来规避此问题。
+> [!warning] NACK requeue 的行为
+>
+> `Nack(false, true)` 会将消息重新放回队列头部。如果消费者立即再次消费，可能导致"毒消息"反复重试。Gen2D 通过 `maxRetry=3` 限制重试次数，并在超过后 ACK 丢弃来规避此问题。
 
-## Metrics 指标
+### Metrics 指标
 
 | Prometheus 指标 | 类型 | Label | 说明 |
 |-----------------|------|-------|------|
@@ -167,7 +174,7 @@ if retry, ok := d.Headers["x-retry-count"].(int32); ok {
 | `gen2d_queue_errors_total` | Counter | `driver=rabbitmq`, `error_type` | 错误总量 |
 | `gen2d_queue_submit_duration_seconds` | Histogram | `driver=rabbitmq` | 发布耗时 |
 
-## 关闭流程
+### 关闭流程
 
 ```go
 func (q *RabbitMQQueue) Close() error {
@@ -181,7 +188,7 @@ func (q *RabbitMQQueue) Close() error {
 
 **关闭顺序**：Channel 先于 Connection 关闭，确保所有未确认的消息被释放回队列。
 
-## 配置参考
+### 配置参考
 
 ```yaml
 # config.yaml
@@ -203,7 +210,7 @@ taskqueue:
 
 ## 关联文档
 
-- [索引](00-index.md) — 文档导航与架构总览图
-- [任务队列](03-task-queue.md) — 可插拔接口与 MemoryQueue 实现
-- [协程池](02-worker-pool.md) — 单机并发控制
-- [系统总览](01-system-overview.md) — 分层架构与配置级联
+- [[00-索引]] — 文档导航与架构总览图
+- [[03-任务队列]] — 可插拔接口与 MemoryQueue 实现
+- [[02-协程池]] — 单机并发控制
+- [[01-系统总览]] — 分层架构与配置级联