Golang基于Redis集群流的高可靠队列客户端插件ami使用指南

什么是ami

ami是一个基于Redis Streams的高可靠消息队列客户端插件，专为Golang设计。它提供了消息的生产、消费、确认等完整功能，特别适合构建分布式系统中的可靠消息队列。

主要特性

• 基于Redis Streams实现，支持Redis集群
• 消息持久化和可靠性保证
• 消费者组支持，实现负载均衡
• 消息重试和死信队列机制
• 简洁易用的API接口

安装

go get github.com/ami-team/ami

基本使用示例

1. 初始化客户端

package main

import (
	"context"
	"fmt"
	"time"

	"github.com/ami-team/ami"
	"github.com/go-redis/redis/v8"
)

func main() {
	// 创建Redis客户端
	redisClient := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs: []string{"127.0.0.1:7000", "127.0.0.1:7001", "127.0.0.1:7002"},
	})

	// 创建ami客户端
	client, err := ami.NewClient(
		ami.WithRedisClient(redisClient),
		ami.WithStreamName("my_stream"),
		ami.WithConsumerGroup("my_group"),
		ami.WithConsumerName("consumer1"),
	)
	if err != nil {
		panic(err)
	}
	defer client.Close()
	
	// 使用示例...
}

2. 生产消息

// 生产消息
msgID, err := client.Produce(context.Background(), &ami.Message{
	Body: []byte("hello world"),
	Headers: map[string]interface{}{
		"key1": "value1",
		"key2": 123,
	},
})
if err != nil {
	fmt.Printf("Produce message failed: %v\n", err)
	return
}
fmt.Printf("Message produced with ID: %s\n", msgID)

3. 消费消息

// 消费消息
messages, err := client.Consume(context.Background(), 10) // 一次最多获取10条消息
if err != nil {
	fmt.Printf("Consume message failed: %v\n", err)
	return
}

for _, msg := range messages {
	fmt.Printf("Received message: ID=%s, Body=%s, Headers=%v\n", 
		msg.ID, string(msg.Body), msg.Headers)
	
	// 处理消息...
	
	// 确认消息处理完成
	if err := client.Ack(context.Background(), msg.ID); err != nil {
		fmt.Printf("Ack message failed: %v\n", err)
		continue
	}
}

高级功能

1. 消费者组管理

// 创建消费者组
err := client.CreateConsumerGroup(context.Background(), "my_group", "$")
if err != nil && !strings.Contains(err.Error(), "BUSYGROUP") {
	// 忽略"消费者组已存在"的错误
	panic(err)
}

// 获取消费者组信息
groups, err := client.ConsumerGroups(context.Background())
if err != nil {
	panic(err)
}
fmt.Printf("Consumer groups: %v\n", groups)

2. 消息重试和死信队列

// 消费消息并处理失败时进行重试
messages, err := client.Consume(context.Background(), 10)
if err != nil {
	panic(err)
}

for _, msg := range messages {
	err := processMessage(msg)
	if err != nil {
		// 处理失败，进行重试
		retryCount := getRetryCount(msg.Headers)
		if retryCount < 3 {
			// 更新重试次数并重新入队
			msg.Headers["retry_count"] = retryCount + 1
			_, err := client.Produce(context.Background(), msg)
			if err != nil {
				fmt.Printf("Retry produce failed: %v\n", err)
			}
		} else {
			// 超过重试次数，进入死信队列
			_, err := client.ProduceToDLQ(context.Background(), msg)
			if err != nil {
				fmt.Printf("Produce to DLQ failed: %v\n", err)
			}
		}
	}
	
	// 无论成功失败都确认消息
	_ = client.Ack(context.Background(), msg.ID)
}

3. 批量生产和消费

// 批量生产消息
msgs := []*ami.Message{
	{Body: []byte("msg1")},
	{Body: []byte("msg2")},
	{Body: []byte("msg3")},
}
msgIDs, err := client.BatchProduce(context.Background(), msgs)
if err != nil {
	panic(err)
}
fmt.Printf("Batch produced message IDs: %v\n", msgIDs)

// 批量确认消息
var ids []string
for _, msg := range messages {
	ids = append(ids, msg.ID)
}
err = client.BatchAck(context.Background(), ids)
if err != nil {
	panic(err)
}

最佳实践

1. 合理设置消费者组：为不同的服务使用不同的消费者组，实现消息的广播

2. 处理Pending消息：定期检查Pending消息，防止消息堆积

pending, err := client.PendingMessages(context.Background())
if err != nil {
	panic(err)
}
fmt.Printf("Pending messages: %v\n", pending)

1. 监控队列长度：监控队列长度，及时发现处理能力不足

length, err := client.StreamLength(context.Background())
if err != nil {
	panic(err)
}
fmt.Printf("Stream length: %d\n", length)

1. 优雅关闭：在服务关闭时确保所有消息处理完成

func gracefulShutdown(client *ami.Client) {
	// 停止接收新消息
	client.StopConsuming()
	
	// 等待处理中的消息完成
	time.Sleep(5 * time.Second)
	
	// 关闭客户端
	client.Close()
}

总结

ami插件为Golang提供了基于Redis Streams的高可靠消息队列解决方案，具有简单易用、高可靠性和高性能的特点。通过合理配置和使用，可以构建出稳定可靠的分布式消息系统。