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

debug

Browse Source
This commit is contained in:
bryanqiu 2023-11-24 11:33:58 +08:00
parent 1e8c7645bb
commit ee6659ccb1
2 changed files with 52 additions and 34 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
go.mod
View File

@ -1,3 +1,3 @@
module qoobing.com/gomod/redis module qoobing.com/gomod/redis
go 1.16 go 1.18

70
xstream/reader.go
View File

@ -1,12 +1,13 @@
package xstream package xstream
import ( import (
"context"
"errors" "errors"
"fmt" "fmt"
"net" "net"
"time" "time"
"github.com/garyburd/redigo/redis" "github.com/gomodule/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"
@ -28,16 +29,34 @@ import (
////// CreateTime time.Time `json:"createtime"` ////// 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) { ////// func runNewAccountMarketingEngine(ctx context.Context) {
////// // r read message from redis ////// // r read message from redis
////// cfg := config.Instance() ////// cfg := config.Instance()
////// r := NewReader(xstreamConf{ ////// r := xstream.NewReader(xstream.Config{
////// group: CORESERVER_NEWACCOUNT_GROUP, ////// Group: CORESERVER_NEWACCOUNT_GROUP,
////// queue: CORESERVER_NEWACCOUNT_QUEUE, ////// Stream: CORESERVER_NEWACCOUNT_QUEUE,
////// sentinelConfig: cfg.CoreserverRedis, ////// Sentinel: cfg.CoreserverRedis,
////// }) ////// })
////// defer r.Close() ////// 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 ////// var error_wait_time = 0 * time.Millisecond
////// for { ////// for {
@ -48,7 +67,7 @@ import (
////// //////
////// // Step 1. fetch message ////// // Step 1. fetch message
////// var err error = nil ////// var err error = nil
////// var msg xstreamMessage ////// var msg xstream.Message
////// var newacctMsg NewAccountMessage ////// var newacctMsg NewAccountMessage
////// log.Infof("waiting fetch newaccount message from redis queue...") ////// log.Infof("waiting fetch newaccount message from redis queue...")
////// if msg, err = r.FetchMessage(ctx); err != nil { ////// if msg, err = r.FetchMessage(ctx); err != nil {
@ -73,7 +92,6 @@ import (
////// // xxxxxxx handle message xxxxxxxxxxxxxxxxxxxxxxxx ////// // 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 ////// // Step 4. send ack to coreserver redis
////// if err := r.CommitMessages(ctx, msg); err != nil { ////// if err := r.CommitMessages(ctx, msg); err != nil {
////// log.Errorf("failed commit message to redis, err:%s", err) ////// log.Errorf("failed commit message to redis, err:%s", err)
@ -89,12 +107,12 @@ var (
) )
type Config struct { type Config struct {
stream string Group string `toml:"group"`
group string Stream string `toml:"stream"`
sentinelConfig sentinel.Config Sentinel sentinel.Config `toml:"sentinel"`
} }
type xstreamMessage struct { type Message struct {
Id string Id string
Value []byte Value []byte
} }
@ -111,9 +129,9 @@ type xstreamReader struct {
func NewReader(cfg Config) (r *xstreamReader) { func NewReader(cfg Config) (r *xstreamReader) {
// redis pool // redis pool
poolId := cfg.sentinelConfig.Sentinels poolId := cfg.Sentinel.Sentinels
if _, ok := redisPools[poolId]; !ok { if _, ok := redisPools[poolId]; !ok {
redisPools[poolId] = sentinel.NewPool(cfg.sentinelConfig) redisPools[poolId] = sentinel.NewPool(cfg.Sentinel)
} }
// initialize // initialize
@ -126,7 +144,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) 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))
} }
@ -219,10 +237,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")
@ -243,7 +261,7 @@ func (r *xstreamReader) fetchMessage() (ret []any, err error) {
return ret, nil return ret, nil
} }
func (r *xstreamReader) FetchMessage() (msg xstreamMessage, err error) { func (r *xstreamReader) FetchMessage(ctx context.Context) (msg Message, 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
@ -289,10 +307,10 @@ func (r *xstreamReader) FetchMessage() (msg xstreamMessage, err error) {
return msg, nil return msg, nil
} }
func (r *xstreamReader) CommitMessages(msg xstreamMessage) (err error) { func (r *xstreamReader) CommitMessages(ctx context.Context, msg Message) (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
} }
@ -319,11 +337,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
} }
@ -331,11 +349,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
} }