上篇我们介绍了RedisStream实现消息队列和发布订阅模式 ，今天我们将介绍Go语言的实现方法。

消息队列

接上篇的例子，我们通过go-redis库的接口XAdd、XReadGroup、XGroupCreateMkStream来实现，详细如下：

package main

import (
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
)
func main() {
	var (
		ctx      = context.Background()
		consumer = "consumer1"
		stream   = "mystream"
		group    = "group1"
		start    = ">"
		count    = 10
	)

	// 创建Redis客户端
	client := redis.NewClient(&redis.Options{
		Addr:     "localhost:6379",
		Password: "",
		DB:       0,
	})
	// 创建消费者组
	err := client.XGroupCreateMkStream(ctx, stream, group, "0-0").Err()
	if err != nil {
		panic(err)
	}
	//添加消息
	_, err = client.XAdd(ctx, &redis.XAddArgs{
		Stream: stream,
		ID:     "*",
		Values: map[string]interface{}{
			"name": "foo",
		},
	}).Result()
	if err != nil {
		panic(err)
	}
	//读取消息
	messages, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
		Group:    group,
		Consumer: consumer,
		Streams:  []string{stream, start},
		Count:    int64(count),
		Block:    0,
		NoAck:    false,
	}).Result()
	if err != nil {
		panic(err)
	} 
	for _, message := range messages[0].Messages {
		fmt.Printf("Received message: %v\n", message.Values)
	}
}

上例实现了一个简单的消息队列，可以实现消息的发送和接收，并保证每条消息只被处理一次 。

阻塞超时

通过修改Block（单位为ms）参数来实现阻塞，详细修改如下

for {
		fmt.Println("Start Receiving", time.Now().String())
		//读取消息
		messages, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
			Group:    group,
			Consumer: consumer,
			Streams:  []string{stream, start},
			Count:    int64(count),
			Block:    2000,
			NoAck:    false,
		}).Result()
		if err != nil {
			panic(err)
		}
		fmt.Println("Received Length=", len(messages), time.Now().String()) 
    ...
	}

输出如下:

Start Receiving 2023-04-08 21:39:23.9669498 +0800 CST m=+0.529310801
Received Length= 1 2023-04-08 21:39:24.0419963 +0800 CST m=+0.604357301
Received message: map[name:foo]
Start Receiving 2023-04-08 21:39:24.0419963 +0800 CST m=+0.604357301
panic: read tcp 127.0.0.1:63777->127.0.0.1:6379: i/o timeout

可见,XReadGroup一直阻塞等待新的消息，直到 Redis 连接超时或者命令被中断。 ps: 如若不需要阻塞等待可修改Block为<0 的值实现。

发布订阅模式

接着，我们实现上篇提到的发布订阅模式，XGroupCreateMkStream创建另外一个订阅组group2，两个协程XReadGroup接收消息，往stream中XAdd消息，两个订阅者将都收到消息，详细如下：

package main

import (
	"context"
	"fmt"
	"github.com/go-redis/redis/v8"
	"time"
)
func main() {
	var (
		ctx       = context.Background()
		consumer  = "consumer1"
		consumer2 = "consumer2"
		stream    = "mystream"
		group     = "group1"
		group2    = "group2"
		start     = ">"
		count     = 10
	)
	// 创建Redis客户端
	client := redis.NewClient(&redis.Options{
		Addr:        "localhost:6379",
		Password:    "",
		DB:          0,
		ReadTimeout: time.Second * 10,
	})
	// 创建消费者组
	err := client.XGroupCreateMkStream(ctx, stream, group, "0-0").Err()
	if err != nil {
		//panic(err)
	}
	// 创建消费者组2
	err = client.XGroupCreateMkStream(ctx, stream, group2, "0-0").Err()
	if err != nil {
		//panic(err)
	}
	//添加消息
	_, err = client.XAdd(ctx, &redis.XAddArgs{
		Stream: stream,
		ID:     "*",
		Values: map[string]interface{}{
			"name": "foo",
		},
	}).Result()
	if err != nil {
		panic(err)
	}
	go func() {
		for {
			fmt.Println("Start Receiving", time.Now().String())
			//读取消息
			messages, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
				Group:    group,
				Consumer: consumer,
				Streams:  []string{stream, start},
				Count:    int64(count),
				Block:    2000,
				NoAck:    false,
			}).Result()
			if err != nil {
				panic(err)
			}
			fmt.Println("Received Length=", len(messages), time.Now().String())
			for _, message := range messages[0].Messages {
				fmt.Printf("Received message: %v\n", message.Values)
			}

		}
	}()
	//读取消息2
	go func() {
		for {
			fmt.Println("2 Start Receiving ", time.Now().String())
			messages, err := client.XReadGroup(ctx, &redis.XReadGroupArgs{
				Group:    group2,
				Consumer: consumer2,
				Streams:  []string{stream, start},
				Count:    int64(count),
				Block:    2000,
				NoAck:    false,
			}).Result()
			if err != nil {
				panic(err)
			}
			fmt.Println("2 Received Length=", len(messages), time.Now().String())
			for _, message := range messages[0].Messages {
				fmt.Printf("2Received message: %v\n", message.Values)
			}
		}
	}()
	for {
	}
}

输出如下：

Start Receiving 2023-04-08 22:19:41.9981051 +0800 CST m=+0.347161101
2 Start Receiving  2023-04-08 22:19:41.9981051 +0800 CST m=+0.347161101
Received Length= 1 2023-04-08 22:19:42.0571709 +0800 CST m=+0.406226901
Received message: map[name:foo]
Start Receiving 2023-04-08 22:19:42.0571709 +0800 CST m=+0.406226901
2 Received Length= 1 2023-04-08 22:19:42.3628412 +0800 CST m=+0.711897201
2Received message: map[name:foo]

参考