go-mq: 声明式配置的RabbitMQ客户端库

go-mq 是一个基于声明式配置的 RabbitMQ 客户端库，它简化了在 Golang 中使用 RabbitMQ 的复杂性，让开发者可以通过配置而非代码来定义队列、交换机和绑定关系。

核心特性

1. 声明式配置：通过结构体标签定义队列、交换机和绑定
2. 自动连接管理：自动处理连接断开和重连
3. 多种消费模式：支持简单消费者、工作队列、发布/订阅等模式
4. 中间件支持：可插入日志、重试、监控等中间件

安装

go get github.com/wx-up/go-mq

基本使用示例

1. 生产者示例

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/wx-up/go-mq"
)

// 定义消息结构体
type OrderMessage struct {
	OrderID   string `json:"order_id"`
	ProductID int    `json:"product_id"`
	Quantity  int    `json:"quantity"`
}

func main() {
	// 创建RabbitMQ客户端
	client, err := mq.New(
		mq.WithAddress("amqp://guest:guest@localhost:5672/"),
		mq.WithLogger(mq.DefaultLogger),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()

	// 创建生产者
	producer := client.NewProducer("order_exchange")

	// 发送消息
	msg := OrderMessage{
		OrderID:   "12345",
		ProductID: 1001,
		Quantity:  2,
	}

	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()

	err = producer.Publish(ctx, msg, 
		mq.WithRoutingKey("order.created"),
		mq.WithHeader("retry", "3"),
	)
	if err != nil {
		fmt.Printf("发送消息失败: %v\n", err)
		return
	}

	fmt.Println("消息发送成功")
}

2. 消费者示例

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/wx-up/go-mq"
)

// 定义消息结构体
type OrderMessage struct {
	OrderID   string `json:"order_id"`
	ProductID int    `json:"product_id"`
	Quantity  int    `json:"quantity"`
}

func main() {
	// 创建RabbitMQ客户端
	client, err := mq.New(
		mq.WithAddress("amqp://guest:guest@localhost:5672/"),
		mq.WithLogger(mq.DefaultLogger),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()

	// 定义消费者配置
	consumerCfg := &mq.ConsumerConfig{
		Exchange: "order_exchange",
		Queue:    "order_queue",
		Routing:  []string{"order.created"},
		Concurrency: 5, // 并发消费者数量
	}

	// 创建消费者
	consumer := client.NewConsumer(consumerCfg)

	// 定义消息处理函数
	handler := func(ctx context.Context, msg *OrderMessage) error {
		fmt.Printf("收到订单消息: %+v\n", msg)
		// 处理订单逻辑...
		return nil
	}

	// 启动消费者
	ctx := context.Background()
	err = consumer.Start(ctx, handler)
	if err != nil {
		fmt.Printf("消费者启动失败: %v\n", err)
		return
	}

	// 保持运行
	select {}
}

高级配置

声明式队列定义

type OrderEvent struct {
	EventType string `json:"event_type"`
	OrderID   string `json:"order_id"`
	
	// 通过结构体标签声明队列属性
	Queue    string `mq:"queue=order_events;durable=true;auto_delete=false"`
	Exchange string `mq:"exchange=order_events;type=topic;durable=true"`
	Routing  string `mq:"routing=order.*"`
}

func main() {
	client, err := mq.New(
		mq.WithAddress("amqp://guest:guest@localhost:5672/"),
		mq.WithDeclare([]mq.Declarer{
			&OrderEvent{}, // 自动声明队列、交换机和绑定
		}),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()
	
	// 使用已声明的队列和交换机...
}

中间件使用

func main() {
	client, err := mq.New(
		mq.WithAddress("amqp://guest:guest@localhost:5672/"),
		mq.WithConsumerMiddlewares(
			// 日志中间件
			mq.LoggingMiddleware(),
			// 重试中间件
			mq.RetryMiddleware(3, 1*time.Second),
			// 自定义中间件
			func(next mq.HandlerFunc) mq.HandlerFunc {
				return func(ctx context.Context, msg interface{}) error {
					start := time.Now()
					err := next(ctx, msg)
					fmt.Printf("处理耗时: %v\n", time.Since(start))
					return err
				}
			},
		),
	)
	if err != nil {
		panic(err)
	}
	
	// 创建消费者...
}

错误处理与监控

func main() {
	// 创建带监控的客户端
	client, err := mq.New(
		mq.WithAddress("amqp://guest:guest@localhost:5672/"),
		mq.WithMetrics(&mq.MetricsConfig{
			Namespace: "myapp",
			Subsystem: "rabbitmq",
		}),
	)
	if err != nil {
		panic(err)
	}
	
	// 错误处理示例
	consumer := client.NewConsumer(&mq.ConsumerConfig{
		Exchange: "order_exchange",
		Queue:    "order_queue",
		ErrorHandler: func(ctx context.Context, err error, msg interface{}) {
			// 自定义错误处理逻辑
			fmt.Printf("处理消息失败: %v, 消息: %+v\n", err, msg)
			// 可以选择重试、记录或发送到死信队列
		},
	})
	
	// ...
}

最佳实践

1. 连接管理：始终使用 defer client.Close() 确保连接关闭
2. 消息序列化：默认使用 JSON，可通过 mq.WithSerializer() 配置其他序列化方式
3. 错误处理：为消费者配置适当的错误处理逻辑
4. 资源声明：在生产环境预先声明所有队列和交换机
5. 监控：集成监控中间件跟踪消息处理情况

go-mq 通过声明式配置简化了 RabbitMQ 的使用，同时保持了足够的灵活性来处理各种消息模式。更多高级用法和配置选项可以参考项目的官方文档。