RabbitRoutine - Golang RabbitMQ 自动重连与发布重试库

RabbitRoutine 是一个轻量级的 Golang 库，用于简化 RabbitMQ 连接管理和消息发布重试机制。下面我将介绍如何使用它来实现可靠的 RabbitMQ 消息传递。

核心特性

1. 自动连接管理（断线重连）
2. 消息发布重试机制
3. 轻量级设计，易于集成
4. 支持消息确认模式

安装

go get github.com/yourusername/rabbitroutine

基本用法

1. 初始化连接

package main

import (
	"fmt"
	"log"
	"time"

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

func main() {
	// 配置连接参数
	config := rabbitroutine.Config{
		URL:           "amqp://guest:guest@localhost:5672/",
		RetryInterval: 5 * time.Second, // 重试间隔
		MaxRetries:    3,               // 最大重试次数
	}

	// 创建连接管理器
	connManager, err := rabbitroutine.NewConnectionManager(config)
	if err != nil {
		log.Fatalf("Failed to create connection manager: %v", err)
	}

	// 启动连接
	go connManager.Start()

	// 等待连接就绪
	<-connManager.Ready()
	fmt.Println("RabbitMQ connection established")
}

2. 发布消息

func publishMessage(connManager *rabbitroutine.ConnectionManager) {
	// 获取通道
	channel, err := connManager.Channel()
	if err != nil {
		log.Printf("Failed to get channel: %v", err)
		return
	}
	defer channel.Close()

	// 声明队列
	_, err = channel.QueueDeclare(
		"test_queue", // 队列名
		true,         // 持久化
		false,        // 自动删除
		false,        // 排他
		false,        // 无等待
		nil,          // 参数
	)
	if err != nil {
		log.Printf("Failed to declare queue: %v", err)
		return
	}

	// 发布消息
	err = connManager.PublishWithRetry(
		"",            // 交换机
		"test_queue",  // 路由键
		false,         // 强制
		false,         // 立即
		amqp.Publishing{
			ContentType: "text/plain",
			Body:        []byte("Hello RabbitMQ!"),
		},
	)
	if err != nil {
		log.Printf("Failed to publish message: %v", err)
	} else {
		log.Println("Message published successfully")
	}
}

3. 消费消息

func consumeMessages(connManager *rabbitroutine.ConnectionManager) {
	// 获取通道
	channel, err := connManager.Channel()
	if err != nil {
		log.Printf("Failed to get channel: %v", err)
		return
	}
	defer channel.Close()

	// 声明队列
	queue, err := channel.QueueDeclare(
		"test_queue", // 队列名
		true,         // 持久化
		false,        // 自动删除
		false,        // 排他
		false,        // 无等待
		nil,          // 参数
	)
	if err != nil {
		log.Printf("Failed to declare queue: %v", err)
		return
	}

	// 消费消息
	msgs, err := channel.Consume(
		queue.Name, // 队列
		"",         // 消费者标签
		false,      // 自动确认
		false,      // 排他
		false,      // 无本地
		false,      // 无等待
		nil,        // 参数
	)
	if err != nil {
		log.Printf("Failed to register consumer: %v", err)
		return
	}

	// 处理消息
	go func() {
		for msg := range msgs {
			log.Printf("Received a message: %s", msg.Body)
			// 手动确认消息
			msg.Ack(false)
		}
	}()
}

高级功能

自定义重试策略

config := rabbitroutine.Config{
	URL: "amqp://guest:guest@localhost:5672/",
	RetryPolicy: func(attempt int) time.Duration {
		// 指数退避策略
		return time.Duration(math.Pow(2, float64(attempt))) * time.Second
	},
	MaxRetries: 5,
}

连接状态监控

// 监听连接状态变化
go func() {
	for status := range connManager.Status() {
		switch status {
		case rabbitroutine.Connected:
			log.Println("Connected to RabbitMQ")
		case rabbitroutine.Disconnected:
			log.Println("Disconnected from RabbitMQ")
		case rabbitroutine.Reconnecting:
			log.Println("Reconnecting to RabbitMQ...")
		}
	}
}()

发布确认模式

// 启用发布确认
channel, err := connManager.Channel()
if err != nil {
	return err
}
defer channel.Close()

err = channel.Confirm(false) // 非阻塞确认模式
if err != nil {
	return err
}

confirm := channel.NotifyPublish(make(chan amqp.Confirmation, 1))

// 发布消息
err = channel.Publish(...)
if err != nil {
	return err
}

// 等待确认
if confirmed := <-confirm; !confirmed.Ack {
	return fmt.Errorf("message not acknowledged by broker")
}

最佳实践

1. 对于关键消息，总是启用持久化队列和消息
2. 使用发布确认模式确保消息到达broker
3. 实现消费者幂等性处理
4. 监控连接状态和重试次数
5. 合理设置重试间隔避免雪崩效应

RabbitRoutine 通过封装常见的连接管理和重试逻辑，使得在 Golang 中使用 RabbitMQ 更加可靠和简单。你可以根据项目需求进一步扩展它的功能。