com.snow.auto_monitor/app/utils/rdbmq/rdbmq.go

package rdbmq

import (
	"context"
	"fmt"
	"time"

	"com.snow.auto_monitor/config"
	goredis "github.com/go-redis/redis/v8"
	"github.com/qit-team/snow-core/redis"
)

// const Nil  = goredis.

func init() {
	//加载配置文件
	conf, err := config.Load("./.env")
	if err != nil {
		fmt.Println(err)
	}
	//注册redis类
	err = redis.Pr.Register(redis.SingletonMain, conf.Redis)
	if err != nil {
		fmt.Println(err)
	}
}

func CreateStreamAndGroup(stream string, group string) (err error) {
	rdb := redis.GetRedis(redis.SingletonMain)
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// 创建stream和group
	err = rdb.XGroupCreateMkStream(ctx, stream, group, "$").Err()

	// XGroupCreate，创建一个消费者组
	// err = rdb.XGroupCreate(ctx, stream, group, "$").Err()
	return
}

func WriteOne(stream string, key string, value interface{}) (err error) {
	rdb := redis.GetRedis(redis.SingletonMain)
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()
	// XADD，添加消息到对尾（这个代码每运行一次就增加一次内容）
	err = rdb.XAdd(ctx, &goredis.XAddArgs{
		Stream:     stream, // 设置流stream的 key，消息队列名
		NoMkStream: false,  //为false，key不存在会新建
		MaxLen:     10000,  //消息队列最大长度，队列长度超过设置最大长度后，旧消息会被删除
		Approx:     false,  //默认false，设为true时，模糊指定stram的长度
		ID:         "*",    //消息ID，* 表示由Redis自动生成
		Values: []interface{}{ //消息队列的内容，键值对形式
			key, value,
		},
		// MinID: "id",//超过设置长度值，丢弃小于MinID消息id
		// Limit: 1000, //限制长度，基本不用
	}).Err()
	return
}

func AckOne(stream string, group string, id string) (err error) {
	rdb := redis.GetRedis(redis.SingletonMain)
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// XACK，确认消息
	err = rdb.XAck(ctx, stream, group, id).Err()
	if err != nil {
		fmt.Println(err)
		return
	}
	_, err = rdb.XDel(ctx, stream, id).Result()
	if err != nil {
		fmt.Println("删除消息失败:", err)
	}
	return
}

func ReadOne(stream string, group string, consumer string) (id string, res map[string]interface{}, err error) {
	rdb := redis.GetRedis(redis.SingletonMain)
	ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond)
	defer cancel()

	// XReadGroup，读取消费者中消息
	readgroupval, err := rdb.XReadGroup(ctx, &goredis.XReadGroupArgs{
		// Streams第二个参数为ID，list of streams and ids, e.g. stream1 stream2 id1 id2
		// id为 >，表示最新未读消息ID，也是未被分配给其他消费者的最新消息
		// id为 0 或其他，表示可以获取已读但未确认的消息。这种情况下BLOCK和NOACK都会忽略
		// id为具体ID，表示获取这个消费者组的pending的历史消息，而不是新消息
		Streams:  []string{stream, ">"},
		Group:    group,    //消费者组名
		Consumer: consumer, // 消费者名
		Count:    1,
		Block:    0,
		NoAck:    true, // true-表示读取消息时确认消息
	}).Result()

	if err != nil {
		return "", nil, err
	}

	return readgroupval[0].Messages[0].ID, readgroupval[0].Messages[0].Values, err
}