gomod/redis
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

base: gomod:master
Branches Tags
gomod:master
gomod:main
gomod:v1.3.4
gomod:v1.3.2
gomod:v1.3.1
gomod:v1.3.0
gomod:v1.2.1
gomod:v1.2.0
gomod:v1.1.2
gomod:v1.1.1
gomod:v1.1.0
gomod:v1.0.4
gomod:v1.0.3
gomod:v1.0.2
gomod:v1.0.1
gomod:v1.0.0
..
compare: gomod:v1.1.1
Branches Tags
gomod:main
gomod:master
gomod:v1.3.4
gomod:v1.3.2
gomod:v1.3.1
gomod:v1.3.0
gomod:v1.2.1
gomod:v1.2.0
gomod:v1.1.2
gomod:v1.1.1
gomod:v1.1.0
gomod:v1.0.4
gomod:v1.0.3
gomod:v1.0.2
gomod:v1.0.1
gomod:v1.0.0

No commits in common. "master" and "v1.1.1" have entirely different histories.
master ... v1.1.1

6 changed files with 98 additions and 372 deletions
Show Stats Expand all files Collapse all files

13
go.mod
View File

@ -1,14 +1,3 @@
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
)
go 1.16

44
redis.go
View File

@ -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
var NewPool = redis.NewPool
var NewSentinelPool = sentinel.NewPool
func newSentinelPool() *redis.Pool {
sntnl := &sentinel.Sentinel{
Addrs: []string{":26379", ":26380", ":26381"},
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
},
}
}

98
redis/redis.go
View File

@ -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
},
}
}

65
redis_test.go
View File

@ -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"))
}

47
sentinel/sentinel.go
View File

@ -16,17 +16,14 @@ import (
type Config struct {
Debug bool `toml:"debug"` //调试开关（会在日志打印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连接用户密码
MasterName string `toml:"mastername"` //REDIS主名称（一个哨兵集群可管理多个REDIS主从结构）
Sentinels string `toml:"sentinels"` //哨兵节点列表，逗号分隔，一般配置三个
SentinelUsername string `toml:"sentinel_username"` //哨兵节点连接用户名，不填默认default
SentinelPassword string `toml:"sentinel_password"` //哨兵节点连接密码，不填认为集群无密码
Wait *bool `toml:"wait"` //当无可用连接时，是否等待
MaxIdle *int `toml:"max_idle"` //最大空闲连接数
MaxActive *int `toml:"max_active"` //最大活跃连接数
IdleTimeout *int `toml:"idle_timeout"` //空闲超时时间
Wait *bool `toml:"wait"`
MaxIdle *int `toml:"max_idle"`
MaxActive *int `toml:"max_active"`
IdleTimeout *int `toml:"idle_timeout"`
}
type Sentinel struct {
@ -54,17 +51,9 @@ 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
masterPassword = cfg.Password
sentinelUsername = cfg.SentinelUsername
sentinelPassword = cfg.SentinelPassword
sntnl = &Sentinel{
Addrs: strings.Split(cfg.Sentinels, ","),
MasterName: masterName,
@ -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"
if masterPassword != "" && masterUsername != "" {
okstr, err = redis.String(c.Do("AUTH", masterUsername, masterPassword))
} else if masterPassword != "" {
okstr, err = redis.String(c.Do("AUTH", 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(
"master(%s) got by name '%s' is not redis master",
masterAddr, masterName)
return nil, err
return nil, fmt.Errorf("%s is not redis master", masterAddr)
}
if cfg.Debug {

187
xstream/reader.go
View File

@ -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 `toml:"stream"`
Sentinel sentinel.Config `toml:"sentinel"`
stream string
group string
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 = func(i int) string { return fmt.Sprintf("%s%03d", prefix, i) }
consumer = fmt.Sprintf("%s-comsumer", group)
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++ {
if _, ok := consumers[consumer(i)]; !ok {
consumer_i = consumer(i)
for i := 0; i < MAX_CONSUMER_NUM; i++ {
name := fmt.Sprintf("%s-%03d", consumer, i)
if _, ok := consumers[name]; !ok {
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 (try lock)
success, err := redis.Int(conn.Do("SETNX", consumer_i, id))
conn.Do("GETEX", consumer_i, "EX", 120) /* for SETNX failed forever */
if err != nil && retry >= 2 {
log.Warningf("redis SETNX(%s,%s) error: %s", consumer_i, id, err)
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
// set to redis
success, err := redis.Int(conn.Do("SETNX", consumer, id))
conn.Do("GETEX", consumer, "EX", 120)
if err != nil {
log.Warningf("redis SETNX(%s,%s) error: %s", consumer, id, err)
panic("redis SETNX error")
}
if success == 0 {
i++
retry++
log.Warningf("consumer name(%s) used by others, retry(%d)", consumer_i, retry)
goto LOOP
log.Warningf("consumer name(%s) used by others", consumer)
panic("consumer name used by others")
}
log.Infof("success to get new consumer-name:[%s]", consumer_i)
return consumer_i
conn.Do("GETEX", consumer, "EX", 120)
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.Group, // group
r.cfg.stream, // key
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)
_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
log.Infof("create group, name: %s", r.cfg.group)
_, 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)
_, err = r.redisDo("XGROUP", "CREATE", r.cfg.Stream, r.cfg.Group, "$", "MKSTREAM")
log.Infof("create group, name: %s", r.cfg.group)
_, err = r.redisDo("XGROUP", "CREATE", r.cfg.stream, r.cfg.group, "$", "MKSTREAM")
return err
}