Compare commits
No commits in common. "master" and "v1.1.1" have entirely different histories.
13
go.mod
13
go.mod
@ -1,14 +1,3 @@
|
|||||||
module qoobing.com/gomod/redis
|
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
|
|
||||||
)
|
|
||||||
|
44
redis.go
44
redis.go
@ -1,11 +1,45 @@
|
|||||||
package redis
|
package redis
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"qoobing.com/gomod/redis/redis"
|
"fmt"
|
||||||
|
"time"
|
||||||
|
|
||||||
|
"github.com/gomodule/redigo/redis"
|
||||||
"qoobing.com/gomod/redis/sentinel"
|
"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
|
||||||
|
},
|
||||||
|
}
|
||||||
|
}
|
||||||
|
@ -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
|
|
||||||
},
|
|
||||||
}
|
|
||||||
}
|
|
@ -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"))
|
|
||||||
|
|
||||||
}
|
|
@ -15,18 +15,15 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type Config struct {
|
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 {
|
type Sentinel struct {
|
||||||
@ -54,18 +51,10 @@ type Sentinel struct {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func NewPool(cfg Config) *redis.Pool {
|
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 (
|
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, ","),
|
Addrs: strings.Split(cfg.Sentinels, ","),
|
||||||
MasterName: masterName,
|
MasterName: masterName,
|
||||||
Dial: func(addr string) (redis.Conn, error) {
|
Dial: func(addr string) (redis.Conn, error) {
|
||||||
@ -74,18 +63,6 @@ func NewPool(cfg Config) *redis.Pool {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
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
|
return c, nil
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
@ -94,6 +71,7 @@ func NewPool(cfg Config) *redis.Pool {
|
|||||||
cfg.Wait = new(bool)
|
cfg.Wait = new(bool)
|
||||||
*cfg.Wait = true
|
*cfg.Wait = true
|
||||||
}
|
}
|
||||||
|
|
||||||
if cfg.IdleTimeout == nil {
|
if cfg.IdleTimeout == nil {
|
||||||
cfg.IdleTimeout = new(int)
|
cfg.IdleTimeout = new(int)
|
||||||
*cfg.IdleTimeout = 300
|
*cfg.IdleTimeout = 300
|
||||||
@ -138,13 +116,7 @@ func NewPool(cfg Config) *redis.Pool {
|
|||||||
return nil, err
|
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 {
|
if err != nil {
|
||||||
return nil, fmt.Errorf("redis master AUTH failed: <%s>", err.Error())
|
return nil, fmt.Errorf("redis master AUTH failed: <%s>", err.Error())
|
||||||
} else if okstr != "OK" {
|
} else if okstr != "OK" {
|
||||||
@ -153,10 +125,7 @@ func NewPool(cfg Config) *redis.Pool {
|
|||||||
|
|
||||||
if !TestRole(c, "master") {
|
if !TestRole(c, "master") {
|
||||||
c.Close()
|
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 {
|
if cfg.Debug {
|
||||||
|
@ -1,118 +1,28 @@
|
|||||||
package xstream
|
package xstream
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"context"
|
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"net"
|
"net"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/gomodule/redigo/redis"
|
"github.com/garyburd/redigo/redis"
|
||||||
"qoobing.com/gomod/log"
|
"qoobing.com/gomod/log"
|
||||||
"qoobing.com/gomod/redis/sentinel"
|
"qoobing.com/gomod/redis/sentinel"
|
||||||
"qoobing.com/gomod/str"
|
"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 (
|
var (
|
||||||
redisPools = map[string]*redis.Pool{}
|
redisPools = map[string]*redis.Pool{}
|
||||||
)
|
)
|
||||||
|
|
||||||
type Config struct {
|
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
|
Id string
|
||||||
Value []byte
|
Value []byte
|
||||||
}
|
}
|
||||||
@ -129,9 +39,9 @@ type xstreamReader struct {
|
|||||||
|
|
||||||
func NewReader(cfg Config) (r *xstreamReader) {
|
func NewReader(cfg Config) (r *xstreamReader) {
|
||||||
// redis pool
|
// redis pool
|
||||||
poolId := cfg.Sentinel.Sentinels
|
poolId := cfg.sentinelConfig.Sentinels
|
||||||
if _, ok := redisPools[poolId]; !ok {
|
if _, ok := redisPools[poolId]; !ok {
|
||||||
redisPools[poolId] = sentinel.NewPool(cfg.Sentinel)
|
redisPools[poolId] = sentinel.NewPool(cfg.sentinelConfig)
|
||||||
}
|
}
|
||||||
|
|
||||||
// initialize
|
// initialize
|
||||||
@ -144,7 +54,7 @@ func NewReader(cfg Config) (r *xstreamReader) {
|
|||||||
restart: true,
|
restart: true,
|
||||||
}
|
}
|
||||||
r.id = str.GetRandomNumString(12)
|
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 {
|
if err := r.initGroup(); err != nil {
|
||||||
panic(fmt.Sprintf("initGroup failed: error:%s", err))
|
panic(fmt.Sprintf("initGroup failed: error:%s", err))
|
||||||
}
|
}
|
||||||
@ -169,18 +79,17 @@ func NewReader(cfg Config) (r *xstreamReader) {
|
|||||||
return r
|
return r
|
||||||
}
|
}
|
||||||
|
|
||||||
func newName(conn redis.Conn, id, group, xstream string) string {
|
func newName(conn redis.Conn, id, group string) string {
|
||||||
var (
|
var (
|
||||||
mkeys = []any{}
|
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{}
|
consumers = map[string]string{}
|
||||||
MAX_CONSUMER_NUM = 30
|
MAX_CONSUMER_NUM = 30
|
||||||
)
|
)
|
||||||
|
|
||||||
// get all name
|
// get all name
|
||||||
for i := 0; i < MAX_CONSUMER_NUM; i++ {
|
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...))
|
ret, err := redis.Strings(conn.Do("MGET", mkeys...))
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@ -189,50 +98,38 @@ func newName(conn redis.Conn, id, group, xstream string) string {
|
|||||||
}
|
}
|
||||||
for i, s := range ret {
|
for i, s := range ret {
|
||||||
if s != "" {
|
if s != "" {
|
||||||
name := consumer(i)
|
name := fmt.Sprintf("%s-%03d", consumer, i)
|
||||||
consumers[name] = s
|
consumers[name] = s
|
||||||
|
log.Infof("name:[%s]", name)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
var (
|
|
||||||
i = 0
|
|
||||||
retry = 0
|
|
||||||
consumer_i = ""
|
|
||||||
)
|
|
||||||
LOOP:
|
|
||||||
// get unused name
|
// 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
|
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 {
|
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() {
|
func (r *xstreamReader) Close() {
|
||||||
@ -250,10 +147,10 @@ func (r *xstreamReader) Close() {
|
|||||||
func (r *xstreamReader) fetchMessage() (ret []any, err error) {
|
func (r *xstreamReader) fetchMessage() (ret []any, err error) {
|
||||||
// func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) {
|
// func (r *xstreamReader) fetchMessage(ctx context.Context) (ret []any, err error) {
|
||||||
args := []interface{}{
|
args := []interface{}{
|
||||||
"GROUP", r.cfg.Group, r.name,
|
"GROUP", r.cfg.group, r.name,
|
||||||
"COUNT", 1,
|
"COUNT", 1,
|
||||||
"BLOCK", 864000000,
|
"BLOCK", 864000000,
|
||||||
"STREAMS", r.cfg.Stream,
|
"STREAMS", r.cfg.stream,
|
||||||
}
|
}
|
||||||
if r.restart {
|
if r.restart {
|
||||||
args = append(args, "0-0")
|
args = append(args, "0-0")
|
||||||
@ -274,7 +171,7 @@ func (r *xstreamReader) fetchMessage() (ret []any, err error) {
|
|||||||
return ret, nil
|
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) {
|
// func (r *xstreamReader) FetchMessage(ctx context.Context) (msg xstreamMessage, err error) {
|
||||||
// Step 1. get result from redis xstream
|
// Step 1. get result from redis xstream
|
||||||
var ret, qmsg, msgs []any = nil, nil, nil
|
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
|
return msg, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
func (r *xstreamReader) CommitMessages(ctx context.Context, msg Message) (err error) {
|
func (r *xstreamReader) CommitMessages(msg xstreamMessage) (err error) {
|
||||||
args := []interface{}{
|
args := []interface{}{
|
||||||
r.cfg.Stream, // key
|
r.cfg.stream, // key
|
||||||
r.cfg.Group, // group
|
r.cfg.group, // group
|
||||||
msg.Id, // id
|
msg.Id, // id
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -350,11 +247,11 @@ func (r *xstreamReader) redisDo(cmd string, args ...any) (reply any, err error)
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (r *xstreamReader) initGroup() 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{})
|
groupInfosInf, _ := groups.([]interface{})
|
||||||
if err != nil || len(groupInfosInf) == 0 {
|
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
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -362,11 +259,11 @@ func (r *xstreamReader) initGroup() error {
|
|||||||
groupInfoInf := groupInfo.([]interface{})
|
groupInfoInf := groupInfo.([]interface{})
|
||||||
groupName := string(groupInfoInf[1].([]uint8))
|
groupName := string(groupInfoInf[1].([]uint8))
|
||||||
log.Infof("group name: %s", groupName)
|
log.Infof("group name: %s", groupName)
|
||||||
if groupName == r.cfg.Group {
|
if groupName == r.cfg.group {
|
||||||
return nil
|
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
|
return err
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user