RabbitMQ与Go实现延迟重试机制

延迟重试机制是消息队列中常见的需求，RabbitMQ可以通过"死信队列"(DLX)功能实现。以下是Go语言实现的方案：

核心思路

1. 正常队列设置TTL和死信交换机
2. 消息过期后自动转入死信队列
3. 消费死信队列实现重试

代码实现

package main

import (
	"fmt"
	"log"
	"time"

	amqp "github.com/rabbitmq/amqp091-go"
)

func failOnError(err error, msg string) {
	if err != nil {
		log.Panicf("%s: %s", msg, err)
	}
}

func main() {
	conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
	failOnError(err, "Failed to connect to RabbitMQ")
	defer conn.Close()

	ch, err := conn.Channel()
	failOnError(err, "Failed to open a channel")
	defer ch.Close()

	// 声明死信交换机和队列
	err = ch.ExchangeDeclare(
		"dlx",   // name
		"topic", // type
		true,    // durable
		false,   // auto-deleted
		false,   // internal
		false,   // no-wait
		nil,     // arguments
	)
	failOnError(err, "Failed to declare a DLX exchange")

	_, err = ch.QueueDeclare(
		"dlq", // name
		true,  // durable
		false, // delete when unused
		false, // exclusive
		false, // no-wait
		nil,   // arguments
	)
	failOnError(err, "Failed to declare a DLQ queue")

	// 绑定死信队列到死信交换机
	err = ch.QueueBind(
		"dlq", // queue name
		"#",   // routing key
		"dlx", // exchange
		false,
		nil,
	)
	failOnError(err, "Failed to bind DLQ")

	// 声明主队列并设置死信参数
	args := amqp.Table{
		"x-dead-letter-exchange":    "dlx",
		"x-dead-letter-routing-key": "dlq",
	}
	_, err = ch.QueueDeclare(
		"work_queue", // name
		true,         // durable
		false,        // delete when unused
		false,        // exclusive
		false,        // no-wait
		args,         // arguments
	)
	failOnError(err, "Failed to declare a work queue")

	// 发布带有TTL的消息
	body := "Hello World!"
	err = ch.Publish(
		"",           // exchange
		"work_queue", // routing key
		false,        // mandatory
		false,        // immediate
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte(body),
			Expiration:  "5000", // 5秒后过期进入死信队列
		})
	failOnError(err, "Failed to publish a message")
	log.Printf(" [x] Sent %s", body)

	// 消费死信队列实现重试
	msgs, err := ch.Consume(
		"dlq",   // queue
		"",      // consumer
		false,   // auto-ack
		false,   // exclusive
		false,   // no-local
		false,   // no-wait
		nil,     // args
	)
	failOnError(err, "Failed to register a consumer")

	go func() {
		for d := range msgs {
			log.Printf(" [x] %s (retry)", d.Body)
			// 处理逻辑...
			time.Sleep(1 * time.Second)
			d.Ack(false) // 手动确认
		}
	}()

	log.Printf(" [*] Waiting for messages. To exit press CTRL+C")
	select {}
}

进阶优化

1. 可以设置重试次数上限，避免无限重试
2. 每次重试可以增加延迟时间(指数退避)
3. 最终失败的消息可以存入另一个队列供人工处理

这种方式利用了RabbitMQ的DLX功能，无需额外组件即可实现延迟重试。