RabbitBus - Golang AMQP 轻量级封装库使用指南

RabbitBus 是一个轻量级的 Golang AMQP (RabbitMQ) 封装库，提供了简单易用的 API 来处理消息队列的发布和消费。下面我将详细介绍其使用方法。

安装

go get github.com/rafaeljesus/rabbus

基本使用

1. 初始化连接

package main

import (
	"log"
	"time"

	"github.com/rafaeljesus/rabbus"
)

func main() {
	rb, err := rabbus.New(
		"amqp://guest:guest@localhost:5672/",
		rabbus.Durable(true),
		rabbus.Attempts(5),
		rabbus.Sleep(time.Second*2),
		rabbus.Timeout(time.Second*10),
	)
	if err != nil {
		log.Fatalf("Failed to init rabbus: %v", err)
	}
	defer rb.Close()
	
	// 其他操作...
}

2. 声明交换机和队列

// 声明交换机
err := rb.DeclareExchange(rabbus.ExchangeConfig{
    Name:       "test_exchange",
    Kind:       "direct",
    Durable:    true,
    AutoDelete: false,
})
if err != nil {
    log.Fatalf("Failed to declare exchange: %v", err)
}

// 声明队列
err = rb.DeclareQueue(rabbus.QueueConfig{
    Name:       "test_queue",
    Durable:    true,
    AutoDelete: false,
})
if err != nil {
    log.Fatalf("Failed to declare queue: %v", err)
}

// 绑定队列到交换机
err = rb.BindQueue(rabbus.BindConfig{
    Exchange:   "test_exchange",
    Queue:      "test_queue",
    RoutingKey: "test_routing_key",
})
if err != nil {
    log.Fatalf("Failed to bind queue: %v", err)
}

3. 发布消息

// 发布消息
msg := rabbus.Message{
    Exchange:     "test_exchange",
    Key:          "test_routing_key",
    Payload:      []byte("Hello RabbitMQ"),
    DeliveryMode: rabbus.Persistent,
}

select {
case rb.Emit() <- msg:
    log.Println("Message published")
case <-time.After(time.Second):
    log.Println("Publish timeout")
}

4. 消费消息

// 消费消息
consumer, err := rb.Consumer("test_queue")
if err != nil {
    log.Fatalf("Failed to create consumer: %v", err)
}

go func() {
    for msg := range consumer {
        log.Printf("Received message: %s", string(msg.Body))
        msg.Ack(false) // 确认消息
    }
}()

// 保持程序运行
select {}

高级特性

1. 重试机制

RabbitBus 内置了重试机制，可以在连接失败时自动重试：

rb, err := rabbus.New(
    "amqp://guest:guest@localhost:5672/",
    rabbus.Attempts(3),          // 最大重试次数
    rabbus.Sleep(time.Second*5), // 重试间隔
)

2. 消息确认模式

// 自动确认模式
consumer, err := rb.Consumer("test_queue", rabbus.AutoAck(true))

// 手动确认模式
consumer, err := rb.Consumer("test_queue")
for msg := range consumer {
    // 处理消息
    if err := process(msg); err != nil {
        msg.Nack(false, true) // 拒绝并重新入队
    } else {
        msg.Ack(false)       // 确认处理完成
    }
}

3. 消息优先级

msg := rabbus.Message{
    Exchange:     "test_exchange",
    Key:          "test_routing_key",
    Payload:      []byte("High priority message"),
    Priority:     5, // 优先级 0-9
}

4. 死信队列配置

// 声明死信队列
err := rb.DeclareQueue(rabbus.QueueConfig{
    Name:       "dlx_queue",
    Durable:    true,
    Arguments: rabbus.QueueArguments{
        "x-message-ttl":          60000, // 60秒TTL
        "x-dead-letter-exchange": "dlx_exchange",
    },
})

完整示例

package main

import (
	"log"
	"time"

	"github.com/rafaeljesus/rabbus"
)

func main() {
	// 初始化连接
	rb, err := rabbus.New(
		"amqp://guest:guest@localhost:5672/",
		rabbus.Durable(true),
		rabbus.Attempts(5),
		rabbus.Sleep(time.Second*2),
	)
	if err != nil {
		log.Fatal(err)
	}
	defer rb.Close()

	// 声明交换机
	if err := rb.DeclareExchange(rabbus.ExchangeConfig{
		Name:    "test_ex",
		Kind:    "direct",
		Durable: true,
	}); err != nil {
		log.Fatal(err)
	}

	// 声明队列
	if err := rb.DeclareQueue(rabbus.QueueConfig{
		Name:    "test_q",
		Durable: true,
	}); err != nil {
		log.Fatal(err)
	}

	// 绑定队列
	if err := rb.BindQueue(rabbus.BindConfig{
		Exchange:   "test_ex",
		Queue:      "test_q",
		RoutingKey: "test_key",
	}); err != nil {
		log.Fatal(err)
	}

	// 发布消息
	go func() {
		for i := 0; i < 10; i++ {
			msg := rabbus.Message{
				Exchange: "test_ex",
				Key:      "test_key",
				Payload:  []byte("message " + string(i)),
			}

			select {
			case rb.Emit() <- msg:
				log.Printf("Published message %d", i)
			case <-time.After(time.Second):
				log.Println("Publish timeout")
			}

			time.Sleep(time.Second)
		}
	}()

	// 消费消息
	consumer, err := rb.Consumer("test_q")
	if err != nil {
		log.Fatal(err)
	}

	for msg := range consumer {
		log.Printf("Received: %s", string(msg.Body))
		msg.Ack(false)
	}
}

总结

RabbitBus 提供了以下主要优势：

1. 简洁的 API 设计
2. 自动重连机制
3. 灵活的配置选项
4. 支持多种消息模式
5. 轻量级无额外依赖

对于需要简单可靠地集成 RabbitMQ 的 Golang 应用来说，RabbitBus 是一个不错的选择。