RabbitMQ 4.x 现代消息队列完全指南 2026：Quorum Queues (Raft) + Streams + 可靠性投递 + 生产调优

1
package main
2

3
import (
4
  "context"
5
  "log"
6
  "time"
7

8
  amqp "github.com/rabbitmq/amqp091-go"
9
)
10

11
func main() {
12
  conn, err := amqp.Dial("amqp://admin:admin123456@localhost:5672/")
13
  if err != nil {
14
    log.Fatalf("Failed to connect to RabbitMQ: %v", err)
15
  }
16
  defer conn.Close()
17

18
  ch, err := conn.Channel()
19
  if err != nil {
20
    log.Fatalf("Failed to open a channel: %v", err)
21
  }
22
  defer ch.Close()
23

24
  // 1. 开启生产者确认模式 (Publisher Confirms)
25
  if err := ch.Confirm(false); err != nil {
26
    log.Fatalf("Failed to enable confirms: %v", err)
27
  }
28

29
  confirms := ch.NotifyPublish(make(chan amqp.Confirmation, 1))
30
  returns := ch.NotifyReturn(make(chan amqp.Return, 1))
31

32
  // 2. 异步监听路由失败消息 (mandatory=true 时触发)
33
  go func() {
34
    for r := range returns {
35
      log.Printf("⚠️ Message returned from exchange: %s, routingKey: %s", r.Exchange, r.RoutingKey)
36
    }
37
  }()
38

39
  // 3. 声明死信交换机与死信队列
40
  _ = ch.ExchangeDeclare("dlx.exchange", "direct", true, false, false, false, nil)
41
  _, _ = ch.QueueDeclare("dlx.queue", true, false, false, false, amqp.Table{
42
    "x-queue-type": "quorum", // 使用现代 Quorum 队列
43
  })
44
  _ = ch.QueueBind("dlx.queue", "dead-letter", "dlx.exchange", false, nil)
45

46
  // 4. 声明主业务 Quorum 队列（配置死信投递规则）
47
  args := amqp.Table{
48
    "x-queue-type":              "quorum",        // 声明为 Quorum 队列
49
    "x-dead-letter-exchange":    "dlx.exchange",  // 死信交换机
50
    "x-dead-letter-routing-key": "dead-letter",   // 死信路由键
51
    "x-delivery-limit":          3,               // 最大投递次数，超限直接死信
52
  }
53
  q, err := ch.QueueDeclare("order.process.queue", true, false, false, false, args)
54
  if err != nil {
55
    log.Fatalf("Failed to declare queue: %v", err)
56
  }
57

58
  // 5. 发送强持久化消息
59
  ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
60
  defer cancel()
61

62
  err = ch.PublishWithContext(ctx,
63
    "",     // 默认直连交换机
64
    q.Name, // 路由至目标队列
65
    true,   // mandatory: 若无法路由则退回
66
    false,
67
    amqp.Publishing{
68
      DeliveryMode: amqp.Persistent, // 核心：磁盘持久化
69
      ContentType:  "application/json",
70
      Body:         []byte(`{"order_id":"ORD-20260907-8899","amount":299.0}`),
71
      MessageId:    "msg_uuid_998822", // 业务唯一 ID 用于幂等去重
72
      Timestamp:    time.Now(),
73
    })
74
  if err != nil {
75
    log.Fatalf("Failed to publish message: %v", err)
76
  }
77

78
  // 6. 等待 Broker 多数派 Raft 落盘确认
79
  if confirmed := <-confirms; confirmed.Ack {
80
    log.Printf("✅ Message confirmed by RabbitMQ Quorum cluster! DeliveryTag: %d", confirmed.DeliveryTag)
81
  } else {
82
    log.Printf("❌ Message rejected/nacked by broker! Need retry.")
83
  }
84
}

1
package main
2

3
import (
4
  "log"
5
  amqp "github.com/rabbitmq/amqp091-go"
6
)
7

8
func main() {
9
  conn, _ := amqp.Dial("amqp://admin:admin123456@localhost:5672/")
10
  defer conn.Close()
11
  ch, _ := conn.Channel()
12
  defer ch.Close()
13

14
  // 限制每个消费者预取消息数量（防止单消费者饥饿或积压内存）
15
  _ = ch.Qos(10, 0, false)
16

17
  // 开启消费：关闭 autoAck 强制手动确认
18
  msgs, err := ch.Consume(
19
    "order.process.queue",
20
    "order-worker-01",
21
    false, // autoAck = false
22
    false,
23
    false,
24
    false,
25
    nil,
26
  )
27
  if err != nil {
28
    log.Fatal(err)
29
  }
30

31
  log.Println("⚡ Worker listening for orders. Press CTRL+C to exit.")
32
  for d := range msgs {
33
    log.Printf("Received message: %s [ID: %s]", d.Body, d.MessageId)
34

35
    // 模拟业务处理
36
    success := processOrder(d.Body, d.MessageId)
37
    if success {
38
      // 成功确认
39
      _ = d.Ack(false)
40
    } else {
41
      // 业务处理失败：进入死信队列或重新入队
42
      // requeue = false 时，根据队列配置自动进入 dlx.queue
43
      _ = d.Nack(false, false)
44
      log.Printf("⚠️ Processing failed, sent to Dead Letter Queue!")
45
    }
46
  }
47
}
48

49
func processOrder(body []byte, messageID string) bool {
50
  // 生产环境应在此查询 Redis: SETNX messageID "PROCESSED" EX 86400
51
  // 若 Key 已存在直接返回 true，实现消费幂等性
52
  return true
53
}