fix:mastername

go.mod

@@ -1,3 +1,14 @@
 module qoobing.com/gomod/redis
 
 go 1.18
+
+require (
+	github.com/gomodule/redigo v1.9.2
+	qoobing.com/gomod/log v1.2.8
+	qoobing.com/gomod/str v1.0.5
+)
+
+require (
+	github.com/tylerb/gls v0.0.0-20150407001822-e606233f194d // indirect
+	github.com/tylerb/is v2.1.4+incompatible // indirect
+)

redis.go

@@ -1,45 +1,11 @@
 package redis
 
 import (
-	"fmt"
+	"qoobing.com/gomod/redis/redis"
-	"time"
-
-	"github.com/gomodule/redigo/redis"
 	"qoobing.com/gomod/redis/sentinel"
 )
 
-func newSentinelPool() *redis.Pool {
+type Config = sentinel.Config
-	sntnl := &sentinel.Sentinel{
+
-		Addrs:      []string{":26379", ":26380", ":26381"},
+var NewPool = redis.NewPool
-		MasterName: "mymaster",
+var NewSentinelPool = sentinel.NewPool
-		Dial: func(addr string) (redis.Conn, error) {
-			timeout := 500 * time.Millisecond
-			c, err := redis.DialTimeout("tcp", addr, timeout, timeout, timeout)
-			if err != nil {
-				return nil, err
-			}
-			return c, nil
-		},
-	}
-	return &redis.Pool{
-		MaxIdle:     3,
-		MaxActive:   64,
-		Wait:        true,
-		IdleTimeout: 240 * time.Second,
-		Dial: func() (redis.Conn, error) {
-			masterAddr, err := sntnl.MasterAddr()
-			if err != nil {
-				return nil, err
-			}
-			c, err := redis.Dial("tcp", masterAddr)
-			if err != nil {
-				return nil, err
-			}
-			if !sentinel.TestRole(c, "master") {
-				c.Close()
-				return nil, fmt.Errorf("%s is not redis master", masterAddr)
-			}
-			return c, nil
-		},
-	}
-}

redis/redis.go

@@ -0,0 +1,98 @@
+package redis
+
+import (
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	"github.com/gomodule/redigo/redis"
+	"qoobing.com/gomod/redis/logging"
+	"qoobing.com/gomod/redis/sentinel"
+)
+
+type Config = sentinel.Config
+
+func NewPool(cfg Config) *redis.Pool {
+	if cfg.Master == "" {
+		panic("config is invalid: Master is empty")
+	} else if cfg.Master != "" && cfg.MasterName != "" {
+		//panic("config is invalid: Master & MasterName must not set both")
+	}
+	var (
+		masterAddr     = cfg.Master
+		masterUsername = cfg.Username
+		masterPassword = cfg.Password
+	)
+	if cfg.Wait == nil {
+		cfg.Wait = new(bool)
+		*cfg.Wait = true
+	}
+
+	if cfg.IdleTimeout == nil {
+		cfg.IdleTimeout = new(int)
+		*cfg.IdleTimeout = 300
+	} else if *cfg.IdleTimeout <= 0 {
+		*cfg.IdleTimeout = 86400 * 365 * 10
+	}
+
+	if cfg.MaxIdle == nil {
+		cfg.MaxIdle = new(int)
+		*cfg.MaxIdle = 100
+	} else if *cfg.MaxIdle < 0 {
+		*cfg.MaxIdle = 100
+	}
+
+	if cfg.MaxActive == nil {
+		cfg.MaxActive = new(int)
+		*cfg.MaxActive = 100
+	} else if *cfg.MaxActive < 0 {
+		*cfg.MaxActive = 100
+	}
+
+	var (
+		logPrefix    = "redis"
+		logStdPrefix = "DBUG "
+		logStdWriter = os.Stdout
+		logStdFlags  = log.Ldate | log.Lmicroseconds | log.Lshortfile
+		logStdLogger = log.New(logStdWriter, logStdPrefix, logStdFlags)
+	)
+	return &redis.Pool{
+		MaxIdle:     *cfg.MaxIdle,
+		MaxActive:   *cfg.MaxActive,
+		Wait:        *cfg.Wait,
+		IdleTimeout: time.Duration(*cfg.IdleTimeout) * time.Second,
+		Dial: func() (redis.Conn, error) {
+			c, err := redis.Dial("tcp", masterAddr)
+			if err != nil {
+				return nil, err
+			}
+
+			var okstr = "OK"
+			if masterPassword != "" && masterUsername != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterUsername, masterPassword))
+			} else if masterPassword != "" {
+				okstr, err = redis.String(c.Do("AUTH", masterPassword))
+			}
+
+			if err != nil {
+				return nil, fmt.Errorf("redis master AUTH failed: <%s>", err.Error())
+			} else if okstr != "OK" {
+				return nil, fmt.Errorf("redis master AUTH failed: <%s>", okstr)
+			}
+
+			//// if !TestRole(c, "master") {
+			//// 	c.Close()
+			//// 	err = fmt.Errorf(
+			//// 		"master(%s) got by name '%s' is not redis master",
+			//// 		masterAddr, masterName)
+			//// 	return nil, err
+			//// }
+
+			if cfg.Debug {
+				c = logging.NewLoggingConn(c, logStdLogger, logPrefix)
+			}
+			return c, nil
+		},
+	}
+}

redis_test.go

@@ -0,0 +1,65 @@
+package redis
+
+import (
+	"fmt"
+	"log"
+	"testing"
+
+	"qoobing.com/gomod/redis/sentinel"
+	"qoobing.com/gomod/redis/xstream"
+)
+
+func TestNewConsumer(t *testing.T) {
+	cfg := xstream.Config{
+		Group:  "group_a",
+		Stream: "testxstream_queue",
+		Sentinel: sentinel.Config{
+			Debug:            true,                     //调试开关（会在日志打印REDIS语句)
+			MasterName:       "walletmaster",           //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
+			Username:         "web3wallet",             //REDIS连接用户名
+			Password:         "web3onthemoon@2023#dev", //REDIS连接用户密码
+			Sentinels:        "sentinel-0.redis:5000",  //哨兵节点列表，逗号分隔，一般配置三个
+			SentinelUsername: "sentinel",               //哨兵节点连接用户名，不填默认default
+			SentinelPassword: "password@mhxzkhl#dev",   //哨兵节点连接密码，不填认为集群无密码
+		},
+	}
+
+	for i := 0; i < 15; i++ {
+		i := i
+		name := fmt.Sprintf("name-%d", i)
+		t.Run(name, func(t *testing.T) {
+			t.Parallel()
+			log.Println("name", name)
+			if r := xstream.NewReader(cfg); r == nil {
+				t.Error("create new reader failed")
+			}
+		})
+	}
+}
+
+func TestNewPool(t *testing.T) {
+	cfg := Config{
+		Debug:            true,                     //调试开关（会在日志打印REDIS语句)
+		Master:           "redis-0.redis:2379",     //REDIS主
+		Username:         "web3wallet",             //REDIS连接用户名
+		Password:         "web3onthemoon@2023#dev", //REDIS连接用户密码
+		Sentinels:        "sentinel-0.redis:5000",  //哨兵节点列表，逗号分隔，一般配置三个
+		SentinelUsername: "sentinel",               //哨兵节点连接用户名，不填默认default
+		SentinelPassword: "password@mhxzkhl#dev",   //哨兵节点连接密码，不填认为集群无密码
+	}
+	pool := NewPool(cfg)
+	if pool == nil {
+		t.Errorf("new pool failed")
+	}
+	rds := pool.Get()
+	if rds == nil {
+		t.Errorf("get redis failed")
+	}
+	if ret, err := rds.Do("SET", "ABC", "123"); err != nil {
+		t.Errorf("get redis failed: %s", err)
+	} else {
+		fmt.Println("ABC", ret)
+	}
+	fmt.Println(rds.Do("GET", "ABC"))
+
+}

sentinel/sentinel.go

@@ -16,7 +16,8 @@ import (
 
 type Config struct {
 	Debug            bool   `toml:"debug"`             //调试开关（会在日志打印REDIS语句)
-	MasterName       string `toml:"mastername"`        //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
+	Master           string `toml:"master"`            //REDIS主host&port 与 MasterName 互斥
+	MasterName       string `toml:"mastername"`        //REDIS主名称,值不为空是启动sentinal模式, 与Master互斥
 	Username         string `toml:"username"`          //REDIS连接用户名
 	Password         string `toml:"password"`          //REDIS连接用户密码
 	Sentinels        string `toml:"sentinels"`         //哨兵节点列表，逗号分隔，一般配置三个
@@ -53,6 +54,11 @@ type Sentinel struct {
 }
 
 func NewPool(cfg Config) *redis.Pool {
+	if cfg.MasterName == "" {
+		panic("config is invalid: MasterName is empty")
+	} else if cfg.Master != "" && cfg.MasterName != "" {
+		//panic("config is invalid: Master & MasterName must not set both")
+	}
 	var (
 		masterName       = cfg.MasterName
 		masterUsername   = cfg.Username
@@ -88,7 +94,6 @@ func NewPool(cfg Config) *redis.Pool {
 		cfg.Wait = new(bool)
 		*cfg.Wait = true
 	}
-
 	if cfg.IdleTimeout == nil {
 		cfg.IdleTimeout = new(int)
 		*cfg.IdleTimeout = 300

xstream/reader.go

@@ -144,7 +144,7 @@ func NewReader(cfg Config) (r *xstreamReader) {
 		restart:  true,
 	}
 	r.id = str.GetRandomNumString(12)
-	r.name = newName(r.conn, r.id, cfg.Group)
+	r.name = newName(r.conn, r.id, cfg.Group, cfg.Stream)
 	if err := r.initGroup(); err != nil {
 		panic(fmt.Sprintf("initGroup failed: error:%s", err))
 	}
@@ -169,17 +169,18 @@ func NewReader(cfg Config) (r *xstreamReader) {
 	return r
 }
 
-func newName(conn redis.Conn, id, group string) string {
+func newName(conn redis.Conn, id, group, xstream string) string {
 	var (
 		mkeys            = []any{}
-		consumer         = fmt.Sprintf("%s-comsumer", group)
+		prefix           = fmt.Sprintf("xstream[%s]-group[%s]-", xstream, group)
+		consumer         = func(i int) string { return fmt.Sprintf("%s%03d", prefix, i) }
 		consumers        = map[string]string{}
 		MAX_CONSUMER_NUM = 30
 	)
 
 	// get all name
 	for i := 0; i < MAX_CONSUMER_NUM; i++ {
-		mkeys = append(mkeys, fmt.Sprintf("%s-%03d", consumer, i))
+		mkeys = append(mkeys, consumer(i))
 	}
 	ret, err := redis.Strings(conn.Do("MGET", mkeys...))
 	if err != nil {
@@ -188,38 +189,50 @@ func newName(conn redis.Conn, id, group string) string {
 	}
 	for i, s := range ret {
 		if s != "" {
-			name := fmt.Sprintf("%s-%03d", consumer, i)
+			name := consumer(i)
 			consumers[name] = s
-			log.Infof("name:[%s]", name)
 		}
 	}
 
+	var (
+		i          = 0
+		retry      = 0
+		consumer_i = ""
+	)
+LOOP:
 	// get unused name
-	for i := 0; i < MAX_CONSUMER_NUM; i++ {
+	for ; i < MAX_CONSUMER_NUM; i++ {
-		name := fmt.Sprintf("%s-%03d", consumer, i)
+		if _, ok := consumers[consumer(i)]; !ok {
-		if _, ok := consumers[name]; !ok {
+			consumer_i = consumer(i)
-			consumer = name
 			break
 		}
-		if i == MAX_CONSUMER_NUM-1 {
+	}
-			panic("too many consumer or something bug")
+	if i == MAX_CONSUMER_NUM {
-
+		panic(fmt.Sprintf("too many consumer(more than %d)???", MAX_CONSUMER_NUM))
-		}
 	}
 
-	// set to redis
+	// set to redis (try lock)
-	success, err := redis.Int(conn.Do("SETNX", consumer, id))
+	success, err := redis.Int(conn.Do("SETNX", consumer_i, id))
-	conn.Do("GETEX", consumer, "EX", 120)
+	conn.Do("GETEX", consumer_i, "EX", 120) /* for SETNX failed forever */
-	if err != nil {
+	if err != nil && retry >= 2 {
-		log.Warningf("redis SETNX(%s,%s) error: %s", consumer, id, err)
+		log.Warningf("redis SETNX(%s,%s) error: %s", consumer_i, id, err)
-		panic("redis SETNX error")
+		panic(fmt.Sprintf("redis SETNX error after try [%d] times", retry))
+	}
+
+	// return or retry if something error
+	if err != nil /*&& retry < 2*/ {
+		retry++
+		log.Warningf("redis SETNX(%s,%s) error: %s, retry(%d)", consumer_i, id, err, retry)
+		goto LOOP
 	}
 	if success == 0 {
-		log.Warningf("consumer name(%s) used by others", consumer)
+		i++
-		panic("consumer name used by others")
+		retry++
+		log.Warningf("consumer name(%s) used by others, retry(%d)", consumer_i, retry)
+		goto LOOP
 	}
-	conn.Do("GETEX", consumer, "EX", 120)
+	log.Infof("success to get new consumer-name:[%s]", consumer_i)
-	return consumer
+	return consumer_i
 }
 
 func (r *xstreamReader) Close() {