go.mod

@@ -1,14 +1,3 @@
 module qoobing.com/gomod/redis
 
-go 1.18
+go 1.16
-
-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,11 +1,45 @@
 package redis
 
 import (
-	"qoobing.com/gomod/redis/redis"
+	"fmt"
+	"time"
+
+	"github.com/gomodule/redigo/redis"
 	"qoobing.com/gomod/redis/sentinel"
 )
 
-type Config = sentinel.Config
+func newSentinelPool() *redis.Pool {
-
+	sntnl := &sentinel.Sentinel{
-var NewPool = redis.NewPool
+		Addrs:      []string{":26379", ":26380", ":26381"},
-var NewSentinelPool = sentinel.NewPool
+		MasterName: "mymaster",
+		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

@@ -1,98 +0,0 @@
-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

@@ -1,65 +0,0 @@
-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

@@ -15,18 +15,15 @@ import (
 )
 
 type Config struct {
-	Debug            bool   `toml:"debug"`             //调试开关（会在日志打印REDIS语句)
+	Debug       bool   `toml:"debug"`      //调试开关（会在日志打印REDIS语句)
-	Master           string `toml:"master"`            //REDIS主host&port 与 MasterName 互斥
+	Username    string `toml:"username"`   //REDIS连接用户名
-	MasterName       string `toml:"mastername"`        //REDIS主名称,值不为空是启动sentinal模式, 与Master互斥
+	Password    string `toml:"password"`   //REDIS连接用户密码
-	Username         string `toml:"username"`          //REDIS连接用户名
+	MasterName  string `toml:"mastername"` //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
-	Password         string `toml:"password"`          //REDIS连接用户密码
+	Sentinels   string `toml:"sentinels"`  //哨兵节点列表，逗号分隔，一般配置三个
-	Sentinels        string `toml:"sentinels"`         //哨兵节点列表，逗号分隔，一般配置三个
+	Wait        *bool  `toml:"wait"`
-	SentinelUsername string `toml:"sentinel_username"` //哨兵节点连接用户名，不填默认default
+	MaxIdle     *int   `toml:"max_idle"`
-	SentinelPassword string `toml:"sentinel_password"` //哨兵节点连接密码，不填认为集群无密码
+	MaxActive   *int   `toml:"max_active"`
-	Wait             *bool  `toml:"wait"`              //当无可用连接时，是否等待
+	IdleTimeout *int   `toml:"idle_timeout"`
-	MaxIdle          *int   `toml:"max_idle"`          //最大空闲连接数
-	MaxActive        *int   `toml:"max_active"`        //最大活跃连接数
-	IdleTimeout      *int   `toml:"idle_timeout"`      //空闲超时时间
 }
 
 type Sentinel struct {
@@ -54,18 +51,10 @@ 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
+		masterName     = cfg.MasterName
-		masterUsername   = cfg.Username
+		masterPassword = cfg.Password
-		masterPassword   = cfg.Password
+		sntnl          = &Sentinel{
-		sentinelUsername = cfg.SentinelUsername
-		sentinelPassword = cfg.SentinelPassword
-		sntnl            = &Sentinel{
 			Addrs:      strings.Split(cfg.Sentinels, ","),
 			MasterName: masterName,
 			Dial: func(addr string) (redis.Conn, error) {
@@ -74,18 +63,6 @@ func NewPool(cfg Config) *redis.Pool {
 				if err != nil {
 					return nil, err
 				}
-				var okstr = "OK"
-				if sentinelPassword != "" && sentinelUsername != "" {
-					okstr, err = redis.String(c.Do("AUTH", sentinelUsername, sentinelPassword))
-				} else if sentinelPassword != "" {
-					okstr, err = redis.String(c.Do("AUTH", sentinelPassword))
-				}
-
-				if err != nil {
-					return nil, fmt.Errorf("redis sentinel AUTH failed: <%s>", err.Error())
-				} else if okstr != "OK" {
-					return nil, fmt.Errorf("redis sentinel AUTH failed: <%s>", okstr)
-				}
 				return c, nil
 			},
 		}
@@ -94,6 +71,7 @@ func NewPool(cfg Config) *redis.Pool {
 		cfg.Wait = new(bool)
 		*cfg.Wait = true
 	}
+
 	if cfg.IdleTimeout == nil {
 		cfg.IdleTimeout = new(int)
 		*cfg.IdleTimeout = 300
@@ -138,13 +116,7 @@ func NewPool(cfg Config) *redis.Pool {
 				return nil, err
 			}
 
-			var okstr = "OK"
+			okstr, err := redis.String(c.Do("AUTH", masterPassword))
-			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" {
@@ -153,10 +125,7 @@ func NewPool(cfg Config) *redis.Pool {
 
 			if !TestRole(c, "master") {
 				c.Close()
-				err = fmt.Errorf(
+				return nil, fmt.Errorf("%s is not redis master", masterAddr)
-					"master(%s) got by name '%s' is not redis master",
-					masterAddr, masterName)
-				return nil, err
 			}
 
 			if cfg.Debug {

xstream/reader.go

@@ -1,118 +1,28 @@
 package xstream
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"net"
 	"time"
 
-	"github.com/gomodule/redigo/redis"
+	"github.com/garyburd/redigo/redis"
 	"qoobing.com/gomod/log"
 	"qoobing.com/gomod/redis/sentinel"
 	"qoobing.com/gomod/str"
 )
 
-//////////////////////////////////////////////////////////////////////////////////
-/////////////////////////  example  //////////////////////////////////////////////
-//////////////////////////////////////////////////////////////////////////////////
-//////  const (
-//////  	CORESERVER_NEWACCOUNT_GROUP = "groupid-coreserver-newaccount"
-//////  	CORESERVER_NEWACCOUNT_QUEUE = "coreserver.newaccount"
-//////  )
-//////
-//////  type NewAccountMessage struct {
-//////  	Sign       string    `json:"sign"`
-//////  	Appid      string    `json:"appid"`
-//////  	Userid     int64     `json:"userid"`
-//////  	Account    string    `json:"account"`
-//////  	CreateTime time.Time `json:"createtime"`
-//////  }
-//////
-//////  func Run(ctx context.Context) {
-//////  	var SLEEP_TIME = 10 * time.Second
-//////  	go func() {
-//////  		for {
-//////  			log.Infof("start run newaccount marketing engine ...")
-//////  			runNewAccountMarketingEngine(ctx)
-//////  			log.Infof("newaccount marketing engine exit unexpected, retry %s later again", SLEEP_TIME.String())
-//////  			time.Sleep(SLEEP_TIME)
-//////  		}
-//////  	}()
-//////  }
-//////
-//////  func runNewAccountMarketingEngine(ctx context.Context) {
-//////  	// r read message from redis
-//////  	cfg := config.Instance()
-//////  	r := xstream.NewReader(xstream.Config{
-//////  		Group:    CORESERVER_NEWACCOUNT_GROUP,
-//////  		Stream:   CORESERVER_NEWACCOUNT_QUEUE,
-//////  		Sentinel: cfg.CoreserverRedis,
-//////  	})
-//////  	defer r.Close()
-//////
-//////  	// w write message to kafka
-//////  	w := &kafka.Writer{
-//////  		Addr:     kafka.TCP(cfg.MarketingKafka.Addresses...),
-//////  		Balancer: &kafka.LeastBytes{},
-//////  	}
-//////  	defer w.Close()
-//////
-//////  	var error_wait_time = 0 * time.Millisecond
-//////  	for {
-//////  		// Step 0. wait a moment if need
-//////  		if error_wait_time > 0 {
-//////  			time.Sleep(error_wait_time)
-//////  		}
-//////
-//////  		// Step 1. fetch message
-//////  		var err error = nil
-//////  		var msg xstream.Message
-//////  		var newacctMsg NewAccountMessage
-//////  		log.Infof("waiting fetch newaccount message from redis queue...")
-//////  		if msg, err = r.FetchMessage(ctx); err != nil {
-//////  			log.Errorf("failed fetch message from redis queue, err:%s", err)
-//////  			//error_wait_time = 1000 * time.Millisecond
-//////  			break
-//////  		}
-//////
-//////          // Step 2. decode message
-//////          if err := json.Unmarshal(msg.Value, &newacctMsg); err != nil {
-//////          	log.Errorf("failed json.unmarshal message[%s], err:%s", string(msg.Value), err)
-//////          	continue
-//////          } else if newacctMsg.Userid == 0 || newacctMsg.Account == "" {
-//////          	log.Errorf("invalid newaccount message, userid=[%d]", newacctMsg.Userid)
-//////          	continue
-//////          }
-//////          log.Infof("fetch newaccount message success, account=[%s]", newacctMsg.Account)
-//////
-//////          // Step 3. handle message
-//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
-//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
-//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
-//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
-//////          // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx
-//////  		// Step 4. send ack to coreserver redis
-//////  		if err := r.CommitMessages(ctx, msg); err != nil {
-//////  			log.Errorf("failed commit message to redis, err:%s", err)
-//////  			error_wait_time = 1000 * time.Millisecond
-//////  			continue
-//////  		}
-//////  	}
-//////  }
-///////////////////////////// end example ////////////////////////////////////////////
-
 var (
 	redisPools = map[string]*redis.Pool{}
 )
 
 type Config struct {
-	Group    string          `toml:"group"`
+	stream         string
-	Stream   string          `toml:"stream"`
+	group          string
-	Sentinel sentinel.Config `toml:"sentinel"`
+	sentinelConfig sentinel.Config
 }
 
-type Message struct {
+type xstreamMessage struct {
 	Id    string
 	Value []byte
 }
@@ -129,9 +39,9 @@ type xstreamReader struct {
 
 func NewReader(cfg Config) (r *xstreamReader) {
 	// redis pool
-	poolId := cfg.Sentinel.Sentinels
+	poolId := cfg.sentinelConfig.Sentinels
 	if _, ok := redisPools[poolId]; !ok {
-		redisPools[poolId] = sentinel.NewPool(cfg.Sentinel)
+		redisPools[poolId] = sentinel.NewPool(cfg.sentinelConfig)
 	}
 
 	// initialize
@@ -144,7 +54,7 @@ func NewReader(cfg Config) (r *xstreamReader) {
 		restart:  true,
 	}
 	r.id = str.GetRandomNumString(12)
-	r.name = newName(r.conn, r.id, cfg.Group, cfg.Stream)
+	r.name = newName(r.conn, r.id, cfg.group)
 	if err := r.initGroup(); err != nil {
 		panic(fmt.Sprintf("initGroup failed: error:%s", err))
 	}
@@ -169,18 +79,17 @@ func NewReader(cfg Config) (r *xstreamReader) {
 	return r
 }
 
-func newName(conn redis.Conn, id, group, xstream string) string {
+func newName(conn redis.Conn, id, group string) string {
 	var (
 		mkeys            = []any{}
-		prefix           = fmt.Sprintf("xstream[%s]-group[%s]-", xstream, group)
+		consumer         = fmt.Sprintf("%s-comsumer", 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, consumer(i))
+		mkeys = append(mkeys, fmt.Sprintf("%s-%03d", consumer, i))
 	}
 	ret, err := redis.Strings(conn.Do("MGET", mkeys...))
 	if err != nil {
@@ -189,50 +98,38 @@ func newName(conn redis.Conn, id, group, xstream string) string {
 	}
 	for i, s := range ret {
 		if s != "" {
-			name := consumer(i)
+			name := fmt.Sprintf("%s-%03d", consumer, i)
 			consumers[name] = s
+			log.Infof("name:[%s]", name)
 		}
 	}
 
-	var (
-		i          = 0
-		retry      = 0
-		consumer_i = ""
-	)
-LOOP:
 	// get unused name
-	for ; i < MAX_CONSUMER_NUM; i++ {
+	for i := 0; i < MAX_CONSUMER_NUM; i++ {
-		if _, ok := consumers[consumer(i)]; !ok {
+		name := fmt.Sprintf("%s-%03d", consumer, i)
-			consumer_i = consumer(i)
+		if _, ok := consumers[name]; !ok {
+			consumer = name
 			break
 		}
-	}
+		if i == MAX_CONSUMER_NUM-1 {
-	if i == MAX_CONSUMER_NUM {
+			panic("too many consumer or something bug")
-		panic(fmt.Sprintf("too many consumer(more than %d)???", MAX_CONSUMER_NUM))
+
+		}
 	}
 
-	// set to redis (try lock)
+	// set to redis
-	success, err := redis.Int(conn.Do("SETNX", consumer_i, id))
+	success, err := redis.Int(conn.Do("SETNX", consumer, id))
-	conn.Do("GETEX", consumer_i, "EX", 120) /* for SETNX failed forever */
+	conn.Do("GETEX", consumer, "EX", 120)
-	if err != nil && retry >= 2 {
+	if err != nil {
-		log.Warningf("redis SETNX(%s,%s) error: %s", consumer_i, id, err)
+		log.Warningf("redis SETNX(%s,%s) error: %s", consumer, id, err)
-		panic(fmt.Sprintf("redis SETNX error after try [%d] times", retry))
+		panic("redis SETNX error")
-	}
-
-	// 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 {
-		i++
+		log.Warningf("consumer name(%s) used by others", consumer)
-		retry++
+		panic("consumer name used by others")
-		log.Warningf("consumer name(%s) used by others, retry(%d)", consumer_i, retry)
-		goto LOOP
 	}
-	log.Infof("success to get new consumer-name:[%s]", consumer_i)
+	conn.Do("GETEX", consumer, "EX", 120)
-	return consumer_i
+	return consumer
 }
 
 func (r *xstreamReader) Close() {
@@ -250,10 +147,10 @@ func (r *xstreamReader) Close() {
 func (r *xstreamReader) fetchMessage() (ret []any, err error) {
 	// func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) {
 	args := []interface{}{
-		"GROUP", r.cfg.Group, r.name,
+		"GROUP", r.cfg.group, r.name,
 		"COUNT", 1,
 		"BLOCK", 864000000,
-		"STREAMS", r.cfg.Stream,
+		"STREAMS", r.cfg.stream,
 	}
 	if r.restart {
 		args = append(args, "0-0")
@@ -274,7 +171,7 @@ func (r *xstreamReader) fetchMessage() (ret []any, err error) {
 	return ret, nil
 }
 
-func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, err error) {
+func (r *xstreamReader) FetchMessage() (msg xstreamMessage, err error) {
 	// func (r *xstreamReader) FetchMessage(ctx context.Context) (msg xstreamMessage, err error) {
 	// Step 1. get result from redis xstream
 	var ret, qmsg, msgs []any = nil, nil, nil
@@ -320,10 +217,10 @@ func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, err erro
 	return msg, nil
 }
 
-func (r *xstreamReader) CommitMessages(ctx context.Context, msg Message) (err error) {
+func (r *xstreamReader) CommitMessages(msg xstreamMessage) (err error) {
 	args := []interface{}{
-		r.cfg.Stream, // key
+		r.cfg.stream, // key
-		r.cfg.Group,  // group
+		r.cfg.group,  // group
 		msg.Id,       // id
 	}
 
@@ -350,11 +247,11 @@ func (r *xstreamReader) redisDo(cmd string, args ...any) (reply any, err error)
 }
 
 func (r *xstreamReader) initGroup() error {
-	groups, err := r.redisDo("XINfO", "GROUPS", r.cfg.Stream)
+	groups, err := r.redisDo("XINfO", "GROUPS", r.cfg.stream)
 	groupInfosInf, _ := groups.([]interface{})
 	if err != nil || len(groupInfosInf) == 0 {
-		log.Infof("create group, name: %s", r.cfg.Group)
+		log.Infof("create group, name: %s", r.cfg.group)
-		_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
+		_, err = r.redisDo("XGROUP", "CREATE", r.cfg.stream, r.cfg.group, "$", "MKSTREAM")
 		return err
 	}
 
@@ -362,11 +259,11 @@ func (r *xstreamReader) initGroup() error {
 		groupInfoInf := groupInfo.([]interface{})
 		groupName := string(groupInfoInf[1].([]uint8))
 		log.Infof("group name: %s", groupName)
-		if groupName == r.cfg.Group {
+		if groupName == r.cfg.group {
 			return nil
 		}
 	}
-	log.Infof("create group, name: %s", r.cfg.Group)
+	log.Infof("create group, name: %s", r.cfg.group)
-	_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
+	_, err = r.redisDo("XGROUP", "CREATE", r.cfg.stream, r.cfg.group, "$", "MKSTREAM")
 	return err
 }