基于Go语言实现对Redis Stream消息队列客户端的封装

1、概述

早期，基于Redis实现轻量化的消息队列有3种实现方式，分别是基于List的LPUSH+BRPOP （BRPOPLPUSH）的实现、PUB/SUB发布订阅模式以及基于Sorted-Set实现方式，但是，这三种模式分别有其相应的缺点。

实现方式	缺点
基于List的LPUSH+BRPOP	做消费者确认ACK比较麻烦，不能保证消费者消费消息后是否成功处理的问题，通常需要维护一个额外的列表，且不支持重复消费和分组消费。
PUB/SUB发布订阅模式	若客户端不在线时发布消息会丢失，且消费者客户端出现消息积压，到一定程度，会被强制断开，导致消息意外丢失，可见PUB/SUB模式不适合做消息存储，消息积压类的业务。
基于Sorted-Set实现	由于集合的特点，不允许重复消息，而且消息ID确定有误的话会导致消息的顺序出错。

Redis5.0中发布的Stream类型，也用来实现典型的消息队列。该Stream类型的出现，几乎满足了消息队列具备的全部内容，包括但不限于：

• 消息ID的序列化生成
• 消息遍历
• 消息的阻塞和非阻塞读取
• 消息的分组消费
• 未完成消息的处理
• 消息队列监控

2、相关命令解释

本文基于Redis6.2版本进行说明，注意不同版本的Redis可能有些命令的部分参数会存在差异，但不影响整体使用。Stream相关的命令主要可以分为2大类，一类是与消息队列相关的命令，另一类是与消费者组相关的命令。
与消息队列相关的命令：

• XADD
• XREAD
• XDEL
• XLEN
• XRANGE
• XREVRANGE
• XTRIM

与消费者组相关的命令：

• XGROUP
• XREADGROUP
• XPENDING
• XACK
• XCLAIM
• XINFO

2.1、XADD

解释：XADD命令用于往某个消息队列中添加消息。

• key：表示消息队列的名称，如果不存在就创建。
• [NOMKSTREAM]：可选参数，表示第一个参数key不存在时不创建。
• [MAXLEN|MINID [=|~] threshold [LIMIT count]] ：可选参数，MAXLEN|MINID表示指定消息队列中消息的最大长度或者是消息ID的最小值。=|~表示设置精确的值或者是大约值，threshold 表示具体设置的值，超过threshold值以后，旧的消息将会被删掉。LIMIT count如果设置了会被当做键值对的形式保存在消息体中的第一个位置，另外设置LIMIT时MAXLEN和MINID只能使用~设置大约值（Redis6.2版本后才加入了LIMIT参数），由于队列中的消息不会主动被删除，但是在设置MAXLEN后，当消息队列长度超过MAXLEN时，会删除老的消息，保证消息队列长度不会一直累加。
• *|ID：表示消息ID，*表示由Redis生成（建议方案），ID表示自己指定。
• field value [field value …]：用于保存消息具体内容的键值对，可以传入多组键值对。

2.2、XREAD

解释：XREAD命令用于从某个消息队列中读取消息，分为阻塞模式和非阻塞模式。

• [COUNT count]：可选参数，COUNT为关键字，表示指定读取消息的数量，count表示具体的值。
• [BLOCK milliseconds]：可选参数，BLOCK为关键字，表示设置XREAD为阻塞模式，默认是非阻塞模式，milliseconds表示具体阻塞的时间。
• STREAMS ：关键字。
• key [key …]：表示消息队列的名称，可以传入多个消息队列名称。
• ID [ID …]：用于表示从哪个消息ID开始读取（不包含），与前面的key一一对应。0表示从第一条消息开始。在阻塞模式中，可以使用$用于表示最新的消息ID。（在非阻塞模式下$无意义）。

2.3、XDEL

解释：XDEL命令用于进行消息删除，注意XACK进行消息确认只是进行了标记，消息还是会存在消息队列中，并没有删除。使用XDEL命令才会将消息从消息队列中删除。

• key：表示消息队列的名称。
• ID [ID …]：表示消息ID，可以传入多个消息ID。

2.4、XLEN

解释：XLEN命令用于获取消息队列的长度。

• key：表示消息队列的名称。

2.5、XRANGE

解释：XRANGE命令用于获取消息队列中的消息，和XREAD有点类似，XREAD只能指定开始消息ID（不包含），XRANGE可以指定开始和结束消息ID。另外还有个XREVRANGE命令用于反向获取消息列表，与XRANGE不同的是消息ID是从大到小。

• key：表示消息队列的名称。
• start：表示起始消息ID（包含）。
• end：表示结束消息ID（包含）。
• [COUNT count]：COUNT为关键字，表示指定读取消息的数量，count表示具体的值。同XREAD命令。

2.6 XTRIM

解释：XTRIM命令用于对消息队列进行修剪，限制长度。

• key：表示消息队列的名称。
• MAXLEN|MINID [=|~] threshold [LIMIT count]：同XADD命令中的同名可选参数意义相同。

2.7、XGROUP

2.7.1 CREATE

解释：XGROUP CREATE命令用于创建消费者组。

• CREATE：关键字，表示创建消费者组命令。
• key：表示消息队列的名称。
• groupname：表示要创建的消费者组名称。
• ID|$：表示该消费者组中的消费者将从哪个消息ID开始消费消息，ID表示指定的消息ID，$表示只消费新产生的消息。
• [MKSTREAM]：可选参数，表示在创建消费者组时，如果指定的消息队列不存在，会自动创建。但是这种方式创建的消息队列其长度为0。

2.7.2 SETID

解释：XGROUP SETID命令用于设置消费者组中下一条要读取的消息ID。

• SETID：关键字，表示设置消费者组中下一条要读取的消息ID命令
• key：表示消息队列的名称。
• groupname：表示消费者组名称。
• ID|$：表示指定具体的消息ID，0可以表示重新开始处理消费者组中的所有消息，$表示只处理消费者组中新产生的消息。

2.7.3 DESTROY

解释：XGROUP DESTROY命令用于销毁消费者组。

• DESTROY：关键字，表示销毁消费者组命令。
• key：表示消息队列的名称。
• groupname：表示要销毁的消费者组名称。

2.7.4 CREATECONSUMER

解释：XGROUP CREATECONSUMER命令用于创建消费者。

• CREATECONSUMER：关键字，表示创建消费者命令。
• key：表示消息队列的名称。
• groupname：表示要创建的消费者所属的消费者组名称。
• consumername：表示要创建的消费者名称。

2.7.5 DELCONSUMER

解释：XGROUP DELCONSUMER命令用于删除消费者。

• DELCONSUMER：关键字，表示删除消费者命令。
• key：表示消息队列的名称。
• groupname：表示要删除的消费者所属的消费者组名称。
• consumername：表示要删除的消费者名称。

2.8、XREADGROUP

解释：XREADGROUP命令用于分组消费消息。

• GROUP：关键字。
• group：表示消费者组名称。
• consumer：表示消费者名称。
• [COUNT count]：可选参数，COUNT为关键字，表示指定读取消息的数量，count表示具体的值。同XREAD命令。
• [BLOCK milliseconds]：可选参数，可选参数，BLOCK为关键字，表示设置XREAD为阻塞模式，默认是非阻塞模式，milliseconds表示具体阻塞的时间。同XREAD命令。
• [NOACK]：可选参数，表示不要将消息加入到PEL队列（Pending等待队列）中，相当于在消息读取时直接进行消息确认。在可靠性要求不高和偶尔丢失消息可接受的场景下可以使用。
• STREAMS：关键字。
• key [key …]：表示消息队列的名称，可以传入多个消息队列名称。同XREAD命令。
• ID [ID …]：用于表示从哪个消息ID开始读取，与前面的key一一对应。0表示从第一条消息开始。在阻塞模式中，可以使用$用于表示最新的消息ID。（在非阻塞模式下$无意义）。同XREAD命令。

2.9、XPENDING

解释：XPENDING命令用于获取等待队列，等待队列中保存的是消费者组内被读取，但是还未完成处理的消息，也就是还没有ACK的消息。

• key：表示消息队列的名称。
• group：表示消费者组名称。
• [IDLE min-idle-time]：可选参数，IDLE表示指定消息已读取时长，min-idle-time表示具体的值。
• start：表示起始消息ID（包含）。
• end：表示结束消息ID（包含）。
• count：指定读取消息的条数。
• [consumer]：可选参数，表示消费者名称。

2.10、XACK

解释：XACK命令用于进行消息确认。

• key：表示消息队列的名称。
• group：表示消费者组名称。
• ID [ID …]：表示消息ID，可以传入多个消息ID。

2.11、XCLAIM：消息转移

解释：XCLAIM命令用于进行消息转移，当某个等待队列中的消息长时间没有被处理（没有ACK）的时候，可以用XCLAIM命令将其转移到其他消费者的等待列表中。

• key：表示消息队列的名称。
• group：表示消费者组名称。
• consumer：表示消费者名称。
• min-idle-time：表示消息空闲时长（表示消息已经读取，但还未处理）。
• ID [ID …]：可选参数，表示要转移的消息的消息ID，可传入多个消息ID。
• [IDLE ms]：可选参数，设置消息空闲时间（上一次读取消息的时间），如果未指定，这假定IDLE为0，即每次转移消息之后重置消息空闲时间。因为如果空闲时间一直累加的话，消息会一直转移。
• [TIME ms-unix-time]：可选参数，与IDLE参数相同，只是它将空闲时间设置为特定的Unix时间（以毫秒为单位），而不是相对的毫秒量。这对于重写生成XCLAIM命令的AOF文件非常有用。
• [RETRYCOUNT count]：可选参数，设置重试计数器的值，每次消息被读取时，该计数器都会递增。一般XCLAIM命令不需要修改重试计数器的值。
• [FORCE]：可选参数，即使指定要转移的消息的消息ID在其他等待列表中不存在，也强制将该消息ID加入到执行消费者的等待列表中。
• [JUSTID]：可选参数，仅返回要转移消息的消息ID，使用此参数意味着重试计数器不会递增。

2.12、XINFO

2.12.1、CONSUMERS

解释：XINFO CONSUMERS命令用于监控消费者。

• CONSUMERS：关键字，表示查看消费者信息命令。
• key：表示消息队列的名称。
• groupname：表示消费者组名称。

2.12.2、GROUPS

解释：XINFO GROUPS命令用于监控消费者组。

• GROUPS：关键字，表示查看消费者组信息命令。
• key：表示消息队列的名称。

2.12.3、STREAM

解释：XINFO STREAM命令用于监控消息队列。

• STREAM：关键字，表示查看消息队列信息命令。
• key：表示消息队列的名称。

2.12.4、HELP

解释：XINFO HELP 命令用于获取帮助。

• HELP：关键字，表示获取帮助信息命令。

3、XADD/XREAD模式和消费者组模式 3.1、XADD/XREAD模式

普通场景下，生产者生产消息，消费者消费消息，多个消费者可以重复的消费相同的消息，比较类似常规的发布订阅模式，订阅了某个消息队列的消费者，能够获取到生产者投放的消息。当然消费者可以采用阻塞或者非阻塞的模式进行读取消息，业务处理等。一个典型的阻塞模式使用方式如下：

#Producer
127.0.0.1:6379> XADD test-mq * key1 value1
"1622605684330-0"
127.0.0.1:6379> 
127.0.0.1:6379> XADD test-mq * key2 value2
"1622605691371-0"
127.0.0.1:6379> 
127.0.0.1:6379> XADD test-mq * key3 value3
"1622605698309-0"
127.0.0.1:6379> 
127.0.0.1:6379> XADD test-mq * key4 value4
"1622605707261-0"
127.0.0.1:6379> 
127.0.0.1:6379> XADD test-mq * key5 value5 key6 value6
"1622605714081-0"
127.0.0.1:6379>

#Consumer
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS test-mq $
1) 1) "test-mq"
   2) 1) 1) "1622605684330-0"
         2) 1) "key1"
            2) "value1"
(3.32s)
127.0.0.1:6379> 
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS test-mq $
1) 1) "test-mq"
   2) 1) 1) "1622605691371-0"
         2) 1) "key2"
            2) "value2"
(2.88s)
127.0.0.1:6379> 
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS test-mq $
1) 1) "test-mq"
   2) 1) 1) "1622605698309-0"
         2) 1) "key3"
            2) "value3"
(3.37s)
127.0.0.1:6379> 
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS test-mq $
1) 1) "test-mq"
   2) 1) 1) "1622605707261-0"
         2) 1) "key4"
            2) "value4"
(3.75s)
127.0.0.1:6379> 
127.0.0.1:6379> XREAD BLOCK 10000 STREAMS test-mq $
1) 1) "test-mq"
   2) 1) 1) "1622605714081-0"
         2) 1) "key5"
            2) "value5"
            3) "key6"
            4) "value6"
(2.47s)
127.0.0.1:6379>

说明：使用阻塞模式的XREAD，XREAD BLOCK 10000 STREAMS test-mq $，最后一个参数$表示读取最新的消息，所以需要先启动消费者，阻塞等待消息，然后生产者添加消息，消费者接受消息完成处理。

3.2、消费者组模式

在有些场景下，我们需要多个消费者配合来消费同一个消息队列中的消息，而不是多个消费者重复的消息，以此来提高消息处理的效率。这种模式也就是消费者组模式了。消费者组模式如下图所示：

下面是Redis Stream的结构图：

上图解释：

• Consumer Group ：消费组，使用 XGROUP CREATE 命令创建，一个消费组有多个消费者(Consumer)。
• last_delivered_id ：游标，每个消费组会有个游标 last_delivered_id，任意一个消费者读取了消息都会使游标 last_delivered_id 往前移动。
• pending_ids ：消费者(Consumer)的状态变量，作用是维护消费者的未确认的 id。 pending_ids 记录了当前已经被客户端读取的消息，但是还没有 ack (Acknowledge character：确认字符）

4、基于Go语言封装Redis Stream客户端Demo

代码结构如下：

connHandle.go

package common

import (
	"fmt"
	"github.com/gomodule/redigo/redis"
	"log"
	"time"
)

func NewClient(opt RedisConnOpt) *RedisStreamMQClient {
	return &RedisStreamMQClient{
		RedisConnOpt: opt,
		ConnPool:     newPool(opt),
	}
}

func newPool(opt RedisConnOpt) *redis.Pool{
	return &redis.Pool{
		MaxIdle: 3,
		IdleTimeout: 240*time.Second,
		MaxActive: 10,
		Wait: true,
		Dial: func() (redis.Conn, error) {
			c, err := redis.Dial("tcp", fmt.Sprintf("%s:%d", opt.Host, opt.Port))
			if err != nil {
				log.Fatalf("Redis.Dial: %v", err)
				return nil, err
			}
			
			if _, err := c.Do("SELECT", opt.Index); err != nil {
				c.Close()
				log.Fatalf("Redis.SELECT: %v", err)
				return nil, err
			}
			return c, nil
		},
	}
}

define.go

package common

import (
	"github.com/gomodule/redigo/redis"
)

const (
	STREAM_MQ_MAX_LEN = 500000  //消息队列最大长度
	READ_MSG_AMOUNT = 1000		//每次读取消息的条数
	READ_MSG_BLOCK_SEC = 30     //阻塞读取消息时间
	TEST_STREAM_KEY = "TestStreamKey1"
)

type RedisConnOpt struct {
	Enable   bool
	Host     string
	Port     int32
	Password string
	Index    int32
	TTL      int32
}

type RedisStreamMQClient struct {
	RedisConnOpt RedisConnOpt
	ConnPool     *redis.Pool
	StreamKey    string		//stream对应的key值
	GroupName    string		//消费者组名称
	ConsumerName string		//消费者名称
}

//等待列表中的消息属性
type PendingMsgInfo struct {
	MsgId			string	//消息ID
	BelongConsumer	string	//所属消费者
	IdleTime		int		//已读取未消费时长
	ReadCount		int		//消息被读取次数
}

// 消息队列信息
type StreamMQInfo struct {
	Length			int64			// 消息队列长度
	RedixTreeKeys	int64			// 基数树key数
	RedixTreeNodes	int64			// 基数树节点数
	LastGeneratedId	string			// 最后一个生成的消息ID
	Groups			int64			// 消费组个数
	FirstEntry		*map[string]map[string]string	// 第一个消息体
	LastEntry		*map[string]map[string]string	// 最后一个消息体
}

// 消费组信息
type GroupInfo struct {
	Name			string	    // 消费组名称
	Consumers		int64		// 组内消费者个数
	Pending			int64		// 组内所有消费者的等待列表总长度
	LastDeliveredId	string		// 组内最后一条被消费的消息ID
}

// 消费者信息
type ConsumerInfo struct {
	Name			string		// 消费者名称
	Pending			int64		// 等待队列长度
	Idle			int64		// 消费者空闲时间（毫秒）
}

msgHandle.go

package common

import (
	"fmt"
	"github.com/gomodule/redigo/redis"
)

// PutMsg 添加消息
func (mqClient *RedisStreamMQClient) PutMsg(streamKey string, msgKey string, msgValue string) (strMsgId string, err error){
	conn := mqClient.ConnPool.Get()
	defer conn.Close()
	/
	startTime := time.Now()
	fmt.Println("Start Test Function GetMsgByGroupConsumer")
	msgMap3, err3 := redisCli.GetMsgByGroupConsumer(common.TEST_STREAM_KEY, groupName, consumerName, msgAmount)

	if err3 != nil {
		fmt.Println("GetMsgByGroupConsumer Failed. err:", err3)
		return
	}
	fmt.Println("End Test Function GetMsgByGroupConsumer")

	fmt.Println("Start Test Function ReplyAck")
	key2msgIds3, _ := PrintMsgMap(msgMap3)
	for streamKey, vecMsgId := range key2msgIds3 {
		//fmt.Println("streamKey:", streamKey, "groupName:", groupName, "consumerName:", consumerName, "Ack Msg Count:", msgCount)
		err3 = redisCli.ReplyAck(streamKey, groupName, vecMsgId)
		if err3 != nil {
			fmt.Println("ReplyAck Failed. err:", err3)
		}
	}

	fmt.Println("End Test Function ReplyAck")
	costTime := time.Since(startTime)
	fmt.Println("=========:START TIME:", startTime)
	fmt.Println("=========:COST TIME:", costTime)
}

// 阻塞消费
func testGroupConsumerBlock(redisCli *common.RedisStreamMQClient, groupName string, consumerName string, msgAmount int32){

	

	startTime := time.Now()
	fmt.Println("Start Test Function GetMsgBlockByGroupConsumer")
	msgMap6, err3 := redisCli.GetMsgBlockByGroupConsumer(common.READ_MSG_BLOCK_SEC,
		common.TEST_STREAM_KEY, groupName, "ConsumerName1-A", msgAmount)

	if err3 != nil {
		fmt.Println("GetMsgBlockByGroupConsumer Failed. err:", err3)
		return
	}
	fmt.Println("End Test Function GetMsgBlockByGroupConsumer")

	fmt.Println("Start Test Function ReplyAck")
	key2msgIds6, msgCount := PrintMsgMap(msgMap6)
	for streamKey, vecMsgId := range key2msgIds6 {
		fmt.Println("streamKey:", streamKey, "groupName:", groupName, "consumerName:", consumerName, "Ack Msg Count:", msgCount)
		err3 = redisCli.ReplyAck(streamKey, groupName, vecMsgId)
		if err3 != nil {
			fmt.Println("ReplyAck Failed. err:", err3)
		}
	}
	fmt.Println("End Test Function ReplyAck")
	costTime := time.Since(startTime)
	fmt.Println("=========:START TIME:", startTime)
	fmt.Println("=========:COST TIME:", costTime)
}

func testPendingList(redisCli *common.RedisStreamMQClient, groupName string, consumerName string){

	fmt.Println("Start Test Function GetPendingList")
	vecPendingMsg, _ := redisCli.GetPendingList(common.TEST_STREAM_KEY, groupName, consumerName, common.READ_MSG_AMOUNT)

	vecMsgId := make([]string, 0)
	for _, pendingMsg := range vecPendingMsg {
		vecMsgId = append(vecMsgId, pendingMsg.MsgId)
	}
	_ = redisCli.ReplyAck(common.TEST_STREAM_KEY, groupName, vecMsgId)
	fmt.Println("Start Test Function GetPendingList")
}

func testXinfo(redisCli *common.RedisStreamMQClient, streamKey string, groupName string) {
	fmt.Println("Start Test Function MonitorMqInfo")
	streamMqInfo := redisCli.MonitorMqInfo(streamKey)
	fmt.Println("streamMqInfo:{")
	fmt.Println("Length:", streamMqInfo.Length)
	fmt.Println("RedixTreeKeys:", streamMqInfo.RedixTreeKeys)
	fmt.Println("RedixTreeNodes:",streamMqInfo.RedixTreeNodes)
	fmt.Println("LastGeneratedId:", streamMqInfo.LastGeneratedId)
	fmt.Println("Groups:",streamMqInfo.Groups)
	fmt.Println("FirstEntry:", streamMqInfo.FirstEntry)
	fmt.Println("LastEntry:", streamMqInfo.LastEntry)
	fmt.Println("}")
	fmt.Println("End Test Function MonitorMqInfo")

	fmt.Println("Start Test Function MonitorConsumerGroupInfo")
	groupInfo := redisCli.MonitorConsumerGroupInfo(streamKey)
	if groupInfo != nil {
		fmt.Println("groupInfo:{")
		fmt.Println("Name:", groupInfo.Name)
		fmt.Println("Consumers:", groupInfo.Consumers)
		fmt.Println("Pending:", groupInfo.Pending)
		fmt.Println("LastDeliveredId:", groupInfo.LastDeliveredId)
		fmt.Println("}")
	}
	fmt.Println("End Test Function MonitorConsumerGroupInfo")

	fmt.Println("Start Test Function MonitorConsumerInfo")
	vecConsumerInfo := redisCli.MonitorConsumerInfo(streamKey, groupName)
	fmt.Println("groupInfo:{")
	for _, consumerInfo := range vecConsumerInfo {
		fmt.Println("Name:", consumerInfo.Name)
		fmt.Println("Pending:", consumerInfo.Pending)
		fmt.Println("Idle:", consumerInfo.Idle)
		fmt.Println("===========================")
	}
	fmt.Println("}")
	fmt.Println("End Test Function MonitorConsumerInfo")
}

func main() {

	fmt.Println("test redis stream mq consumer")
	redisOpt := common.RedisConnOpt{
		Enable: true,
		Host:   "127.0.0.1",
		Port:   6379,
		TTL:    240,
	}

	redisCli := common.NewClient(redisOpt)
	fmt.Println("Test Redis Consumer Client Host:", redisCli.RedisConnOpt.Host,
		", Port:", redisCli.RedisConnOpt.Port, ", DB:", redisCli.RedisConnOpt.Index)

	//无消费者组，所有消费者都能消费所有消息
	//testNoGroup(redisCli)

	//无消费者组，所有消费者都能消费所有消息(阻塞模式)
	//testNoGroupBlock(redisCli)

	//有消费者组，所有消费者都不能重复消费组内的消息
	var groupName1 string = "GroupName1"
	var consumerName1 string = "ConsumerName1-" + time.Now().String()
	var msgAmount1 int32 = 50000
	testGroupConsumer(redisCli, groupName1, consumerName1, msgAmount1)
	testPendingList(redisCli, groupName1, consumerName1)

	//有消费者组，所有消费者都不能重复消费组内的消息(阻塞模式)
	//var groupName2 string = "GroupName2"
	//var consumerName2 string = "ConsumerName2-" + time.Now().String()
	//var msgAmount2 int32 = 50000
	//testGroupConsumerBlock(redisCli, groupName2, consumerName2, msgAmount2)
	//testPendingList(redisCli, groupName2, consumerName2)

	//XINFO测试
	//var groupName1 string = "testgroupname"
	//var streamKey string = "test-mq"
	//testXinfo(redisCli, streamKey, groupName1)

	return
}

这里给出完整代码下载地址，下载可直接运行。