feat: 支持 Redis Stream 消息队列 (#9)

* feat: 支持 Redis Stream 消息队列

* 优化代码取消注释及错误返回

* fix: 修复部分问题

* fix: 修复 Redis 队列不支持多Consumer 问题并优化 Pending 消息取值流程

* fix: 从外部传入 redis 实例，支持集群模式

* fix: 依赖 redis 从 v9 降级到 v8

* fix: 随机生成消费者名称

* fix: 传入 Redis Options 初始化, 使用 v9 库, 支持传入 global_idle, 私有函数名

* fix: 去掉了 redisService 接口, 支持部分 Redis Options 参数

* fix: 增加了连接池大小参数

---------

Co-authored-by: Xudong Liu <xudong.liu@ewp-group.com>

Author: sincerefly

cloud/common.go

@@ -13,17 +13,21 @@ import (
 type Provider string
 
 const (
-	AWSProvider          Provider = "aws"
-	TencentCloudProvider Provider = "tencentcloud"
+	AWSProvider             Provider = "aws"
+	TencentCloudProvider    Provider = "tencentcloud"
+	StandaloneRedisProvider Provider = "standalone_redis"
+	ClusterRedisProvider    Provider = "cluster_redis"
 )
 
 var (
-	ErrUnsupportedCloudProvider = fmt.Errorf("unsupported provider, only support %s and %s", AWSProvider, TencentCloudProvider)
-	ErrProviderNotTencentCloud  = errors.New("provider is not tencentcloud")
-	ErrProviderNotAWS           = errors.New("provider is not aws")
-	ErrEmptySecretID            = errors.New("secret_id is empty")
-	ErrEmptySecretKey           = errors.New("secret_key is empty")
-	ErrEmptyRegion              = errors.New("region is empty")
+	ErrUnsupportedCloudProvider   = fmt.Errorf("unsupported provider, only support %s, %s and %s", AWSProvider, TencentCloudProvider, StandaloneRedisProvider)
+	ErrProviderNotTencentCloud    = errors.New("provider is not tencentcloud")
+	ErrProviderNotAWS             = errors.New("provider is not aws")
+	ErrProviderNotStandaloneRedis = errors.New("provider is not standalone redis")
+	ErrProviderNotClusterRedis    = errors.New("provider is not cluster redis")
+	ErrEmptySecretID              = errors.New("secret_id is empty")
+	ErrEmptySecretKey             = errors.New("secret_key is empty")
+	ErrEmptyRegion                = errors.New("region is empty")
 )
 
 type Option interface {
@@ -35,6 +39,8 @@ type Option interface {
 	GetAssumeRegion() string
 	CheckAWS() error
 	CheckTencentCloud() error
+	CheckStandaloneRedis() error
+	CheckClusterRedis() error
 }
 
 type CommonOption struct {
@@ -104,6 +110,14 @@ func (option CommonOption) CheckTencentCloud() error {
 	return option.check()
 }
 
+func (option CommonOption) CheckStandaloneRedis() error {
+	return ErrProviderNotStandaloneRedis
+}
+
+func (option CommonOption) CheckClusterRedis() error {
+	return ErrProviderNotClusterRedis
+}
+
 // AwsNewSession
 func AwsNewSession(option Option) (*session.Session, *aws.Config, error) {
 	var creds *credentials.Credentials

cloud/examples/queue/queue.go

@@ -9,9 +9,36 @@ import (
 	"github.com/byte-power/gorich/cloud/queue"
 )
 
+// Configure Addr/Addrs, Password to run standalone/cluster redis example.
 // Configure token, url, topic_name and subscription_name to run tencentcloud example.
 // Configure secret_id, secret_key, region, and queue_name to run this example.
 func main() {
+
+	// Redis 单节点
+	//optionForBaseRedis := queue.StandaloneRedisQueueOption{
+	//	Addr:              "localhost:6379",
+	//	Password:          "",
+	//	ConsumerGroup: "save_task_consumer_group",
+	//	Idle:              10,
+	//}
+
+	// Redis 集群
+	optionForBaseRedis := queue.ClusterRedisQueueOption{
+		Addrs: []string{
+			"localhost:7000",
+			"localhost:7001",
+			"localhost:7002",
+			"localhost:7003",
+			"localhost:7004",
+			"localhost:7005",
+		},
+		Password:      "",
+		ConsumerGroup: "save_task_consumer_group",
+		Idle:          10,
+	}
+
+	queue_examples("test_queue_name", optionForBaseRedis)
+
 	optionForTencentCloud := queue.TencentCloudQueueOption{
 		Token: "access_jwt_token_xxx",
 		URL:   "http://pulsar-xxxxxxxxx.tdmq.ap-gz.public.tencenttdmq.com:8080",