cs-note/hzh/Gen2D/04-RabbitMQ集成.md

04 - RabbitMQ 集成

一句话概括：基于 AMQP 的持久化消息队列，支持手动 ACK、失败重试和死信丢弃。

消息流

graph TB
    subgraph Producer["生产者"]
        HANDLER["Handler.Generate()"]
    end

    subgraph RabbitMQ["RabbitMQ"]
        EXCHANGE["Default Exchange<br/>(Direct)"]
        QUEUE["gen2d:tasks<br/>durable=true"]
    end

    subgraph Consumer["消费者"]
        CONSUME["channel.Consume<br/>autoAck=false"]
    end

    subgraph Decision["ACK/NACK 决策树"]
        SUCCESS{"handler 成功?"}
        RETRY{"retry < maxRetry?"}
        ACK_OK["ACK<br/>确认消费"]
        NACK["NACK + requeue<br/>重新入队"]
        ACK_DISCARD["ACK (discard)<br/>丢弃死信"]
    end

    HANDLER -->|"Publish<br/>Persistent"| EXCHANGE
    EXCHANGE --> QUEUE
    QUEUE --> CONSUME
    CONSUME --> SUCCESS
    SUCCESS -->|是| ACK_OK
    SUCCESS -->|否| RETRY
    RETRY -->|是| NACK
    RETRY -->|否| ACK_DISCARD
    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}

参数说明：

参数	默认值	说明
URL	amqp://guest:guest@localhost:5672/	AMQP 连接地址
Queue	gen2d:tasks	队列名称
Prefetch	1	每次预取消息数，1 保证公平调度
MaxRetry	3	失败最大重试次数

💡 Prefetch=1 的含义：每个 Consumer 同时只处理 1 条消息，处理完（ACK）后才接收下一条。这避免了消息堆积在 Consumer 端，配合协程池的并发控制实现精确的任务调度。

消息发布 (Submit)

func (q *RabbitMQQueue) Submit(ctx context.Context, msg TaskMessage) error {
    body, _ := json.Marshal(msg)
    return q.channel.PublishWithContext(ctx,
        "",        // exchange（默认直连）
        q.queue,   // routing key = queue name
        false,     // mandatory
        false,     // immediate
        amqp.Publishing{
            ContentType:  "application/json",
            DeliveryMode: amqp.Persistent,  // 持久化消息
            Body:         body,
            Timestamp:    time.Now(),
            Headers: amqp.Table{
                "x-retry-count": msg.RetryCount,
            },
        },
    )
}

关键配置：

属性	值	说明
DeliveryMode	Persistent (2)	消息写磁盘，RabbitMQ 重启不丢失
ContentType	application/json	JSON 序列化
x-retry-count	int (header)	当前重试次数，供消费端判断

消息消费 (Consume)

func (q *RabbitMQQueue) Consume(ctx context.Context, handler func(TaskMessage) error) error {
    deliveries, _ := q.channel.Consume(
        q.queue,   // queue
        "",        // consumer name（自动生成）
        false,     // autoAck = false（手动 ACK）
        false,     // exclusive
        false,     // noLocal
        false,     // noWait
        nil,       // args
    )
    // 持续消费循环...
}

手动 ACK 模式：

手动 ACK 给予消费者完全的控制权——只有当消息被成功处理后才确认，否则可以选择重试或丢弃。

ACK/NACK 决策树

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/>超过最大重试，记录死信日志"]

场景	操作	说明
handler 成功	d.Ack(false)	确认消费，消息从队列移除
handler 失败 + retry < max	d.Nack(false, true)	拒绝并重新入队，retry count 递增
handler 失败 + retry >= max	d.Ack(false) + 日志	超过最大重试，丢弃（可扩展为死信队列）
JSON 解析失败	d.Ack(false)	格式错误无法恢复，直接丢弃

重试计数传递：

// 发布时：写入 header
Headers: amqp.Table{
    "x-retry-count": msg.RetryCount,
}

// 消费时：从 header 读取
if retry, ok := d.Headers["x-retry-count"].(int32); ok {
    msg.RetryCount = int(retry)
}

⚠️ NACK requeue 的行为：Nack(false, true) 会将消息重新放回队列头部。如果消费者立即再次消费，可能导致"毒消息"反复重试。Gen2D 通过 maxRetry=3 限制重试次数，并在超过后 ACK 丢弃来规避此问题。

Metrics 指标

Prometheus 指标	类型	Label	说明
gen2d_rabbitmq_connection_status	Gauge	—	连接状态（1=connected, 0=disconnected）
gen2d_queue_submitted_total	Counter	driver=rabbitmq	发布消息总量
gen2d_queue_consumed_total	Counter	driver=rabbitmq	成功消费总量
gen2d_queue_errors_total	Counter	driver=rabbitmq, error_type	错误总量
gen2d_queue_submit_duration_seconds	Histogram	driver=rabbitmq	发布耗时

关闭流程

func (q *RabbitMQQueue) Close() error {
    metrics.RabbitMQConnectionStatus.Set(0)  // 标记断开
    var errs []error
    errs = append(errs, q.channel.Close())   // 先关 channel
    errs = append(errs, q.conn.Close())      // 再关连接
    return errors.Join(errs...)
}

关闭顺序：Channel 先于 Connection 关闭，确保所有未确认的消息被释放回队列。

配置参考

# config.yaml
taskqueue:
  driver: "rabbitmq"
  rabbitmq:
    url: "amqp://guest:guest@localhost:5672/"
    queue: "gen2d:tasks"
    prefetch: 1
    max_retry: 3

配置项	环境变量	默认值	说明
url	GEN2D_TASKQUEUE_RABBITMQ_URL	amqp://guest:guest@localhost:5672/	AMQP 连接地址
queue	GEN2D_TASKQUEUE_RABBITMQ_QUEUE	gen2d:tasks	队列名称
prefetch	GEN2D_TASKQUEUE_RABBITMQ_PREFETCH	1	预取消息数
max_retry	GEN2D_TASKQUEUE_RABBITMQ_MAX_RETRY	3	最大重试次数

关联文档

• 索引 — 文档导航与架构总览图
• 任务队列 — 可插拔接口与 MemoryQueue 实现
• 协程池 — 单机并发控制
• 系统总览 — 分层架构与配置级联