diff --git a/README.md b/README.md index 63e37ed..15cf36e 100644 --- a/README.md +++ b/README.md @@ -16,7 +16,8 @@ Core Advantages: - Auto retry failed messages - Works out of the box, Config Nothing and Deploy Nothing, A Redis is all you need. - Natively adapted to the distributed environment, messages processed concurrently on multiple machines -. workers can be added, removed or migrated at any time +. Workers can be added, removed or migrated at any time +- Support Redis Cluster for high availability ## Install @@ -26,8 +27,6 @@ DelayQueue requires a Go version with modules support. Run following command lin go get github.com/hdt3213/delayqueue ``` -> if you are using github.com/go-redis/redis/v8 please use `go get github.com/hdt3213/delayqueue@v8` - ## Get Started ```go @@ -69,6 +68,9 @@ func main() { } ``` +> if you are using github.com/go-redis/redis/v8 please use `go get github.com/hdt3213/delayqueue@v8` +> If you are using redis client other than go-redis, you could wrap your redis client into [RedisCli](https://pkg.go.dev/github.com/hdt3213/delayqueue#RedisCli) interface + ## Options ```go @@ -125,7 +127,38 @@ WithDefaultRetryCount customizes the max number of retry, it effects of messages use WithRetryCount during DelayQueue.SendScheduleMsg or DelayQueue.SendDelayMsg to specific retry count of particular message -# More Details +## Cluster + +If you are using Redis Cluster, please use `NewQueueOnCluster` + +```go +redisCli := redis.NewClusterClient(&redis.ClusterOptions{ + Addrs: []string{ + "127.0.0.1:7000", + "127.0.0.1:7001", + "127.0.0.1:7002", + }, +}) +callback := func(s string) bool { + return true +} +queue := NewQueueOnCluster("test", redisCli, callback) +``` + +If you are using transparent clusters, such as codis, twemproxy, or the redis of cluster architecture on aliyun, tencentcloud, +just use `NewQueue` and enable hash tag + +```go +redisCli := redis.NewClient(&redis.Options{ + Addr: "127.0.0.1:6379", +}) +callback := func(s string) bool { + return true +} +queue := delayqueue.NewQueue("example", redisCli, callback, UseHashTagKey()) +``` + +## More Details Here is the complete flowchart: diff --git a/README_CN.md b/README_CN.md index 2eb0f80..0bf9f6a 100644 --- a/README_CN.md +++ b/README_CN.md @@ -12,6 +12,7 @@ DelayQueue 的主要优势： - 自动重试处理失败的消息 - 开箱即用, 无需部署或安装中间件, 只需要一个 Redis 即可工作 - 原生适配分布式环境, 可在多台机器上并发的处理消息. 可以随时增加、减少或迁移 Worker +- 支持各类 Redis 集群 # 安装 @@ -64,6 +65,9 @@ func main() { } ``` +> 如果您仍在使用 redis/v8 请使用 v8 分支: `go get github.com/hdt3213/delayqueue@v8` +> 如果您在使用其他的 redis 客户端, 可以将其包装到 [RedisCli](https://pkg.go.dev/github.com/hdt3213/delayqueue#RedisCli) 接口中 + # 选项 ```go @@ -117,6 +121,36 @@ WithDefaultRetryCount(count uint) 在调用 DelayQueue.SendScheduleMsg or DelayQueue.SendDelayMsg 发送消息时，可以调用 WithRetryCount 为这条消息单独指定重试次数。 +# 集群 + +如果需要在 Redis Cluster 上工作, 请使用 `NewQueueOnCluster`: + +```go +redisCli := redis.NewClusterClient(&redis.ClusterOptions{ + Addrs: []string{ + "127.0.0.1:7000", + "127.0.0.1:7001", + "127.0.0.1:7002", + }, +}) +callback := func(s string) bool { + return true +} +queue := NewQueueOnCluster("test", redisCli, callback) +``` + +如果是阿里云,腾讯云的 Redis 集群版或 codis, twemproxy 这类透明式的集群, 使用 `NewQueue` 并启用 UseHashTagKey() 即可: + +```go +redisCli := redis.NewClient(&redis.Options{ + Addr: "127.0.0.1:6379", +}) +callback := func(s string) bool { + return true +} +queue := delayqueue.NewQueue("example", redisCli, callback, UseHashTagKey()) +``` + # 更多细节 完整流程如图所示： diff --git a/delayqueue.go b/delayqueue.go index e763026..c54e624 100644 --- a/delayqueue.go +++ b/delayqueue.go @@ -1,11 +1,11 @@ package delayqueue import ( - "context" + "errors" "fmt" "github.com/google/uuid" - "github.com/redis/go-redis/v9" "log" + "strconv" "sync" "time" ) @@ -14,7 +14,7 @@ import ( type DelayQueue struct { // name for this Queue. Make sure the name is unique in redis database name string - redisCli *redis.Client + redisCli RedisCli cb func(string) bool pendingKey string // sorted set: message id -> delivery time readyKey string // list @@ -22,6 +22,7 @@ type DelayQueue struct { retryKey string // list retryCountKey string // hash: message id -> remain retry count garbageKey string // set: message id + useHashTag bool ticker *time.Ticker logger *log.Logger close chan struct{} @@ -31,8 +32,24 @@ type DelayQueue struct { defaultRetryCount uint fetchInterval time.Duration fetchLimit uint + concurrent uint +} - concurrent uint +// NilErr represents redis nil +var NilErr = errors.New("nil") + +// RedisCli is abstraction for redis client, required commands only not all commands +type RedisCli interface { + Eval(script string, keys []string, args []interface{}) (interface{}, error) // args should be string, integer or float + Set(key string, value string, expiration time.Duration) error + Get(key string) (string, error) + Del(keys []string) error + HSet(key string, field string, value string) error + HDel(key string, fields []string) error + SMembers(key string) ([]string, error) + SRem(key string, members []string) error + ZAdd(key string, values map[string]float64) error + ZRem(key string, fields []string) error } type hashTagKeyOpt int @@ -45,9 +62,9 @@ func UseHashTagKey() interface{} { return hashTagKeyOpt(1) } -// NewQueue creates a new queue, use DelayQueue.StartConsume to consume or DelayQueue.SendScheduleMsg to publish message +// NewQueue0 creates a new queue, use DelayQueue.StartConsume to consume or DelayQueue.SendScheduleMsg to publish message // callback returns true to confirm successful consumption. If callback returns false or not return within maxConsumeDuration, DelayQueue will re-deliver this message -func NewQueue(name string, cli *redis.Client, callback func(string) bool, opts ...interface{}) *DelayQueue { +func NewQueue0(name string, cli RedisCli, callback func(string) bool, opts ...interface{}) *DelayQueue { if name == "" { panic("name is required") } @@ -80,6 +97,7 @@ func NewQueue(name string, cli *redis.Client, callback func(string) bool, opts . retryKey: keyPrefix + ":retry", retryCountKey: keyPrefix + ":retry:cnt", garbageKey: keyPrefix + ":garbage", + useHashTag: useHashTag, close: make(chan struct{}, 1), maxConsumeDuration: 5 * time.Second, msgTTL: time.Hour, @@ -132,6 +150,9 @@ func (q *DelayQueue) WithDefaultRetryCount(count uint) *DelayQueue { } func (q *DelayQueue) genMsgKey(idStr string) string { + if q.useHashTag { + return "{dp:" + q.name + "}" + ":msg:" + idStr + } return "dp:" + q.name + ":msg:" + idStr } @@ -165,21 +186,20 @@ func (q *DelayQueue) SendScheduleMsg(payload string, t time.Time, opts ...interf } // generate id idStr := uuid.Must(uuid.NewRandom()).String() - ctx := context.Background() now := time.Now() // store msg msgTTL := t.Sub(now) + q.msgTTL // delivery + q.msgTTL - err := q.redisCli.Set(ctx, q.genMsgKey(idStr), payload, msgTTL).Err() + err := q.redisCli.Set(q.genMsgKey(idStr), payload, msgTTL) if err != nil { return fmt.Errorf("store msg failed: %v", err) } // store retry count - err = q.redisCli.HSet(ctx, q.retryCountKey, idStr, retryCount).Err() + err = q.redisCli.HSet(q.retryCountKey, idStr, strconv.Itoa(int(retryCount))) if err != nil { return fmt.Errorf("store retry count failed: %v", err) } // put to pending - err = q.redisCli.ZAdd(ctx, q.pendingKey, redis.Z{Score: float64(t.Unix()), Member: idStr}).Err() + err = q.redisCli.ZAdd(q.pendingKey, map[string]float64{idStr: float64(t.Unix())}) if err != nil { return fmt.Errorf("push to pending failed: %v", err) } @@ -214,10 +234,9 @@ redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1]) -- remove msgs from pendi func (q *DelayQueue) pending2Ready() error { now := time.Now().Unix() - ctx := context.Background() keys := []string{q.pendingKey, q.readyKey} - err := q.redisCli.Eval(ctx, pending2ReadyScript, keys, now).Err() - if err != nil && err != redis.Nil { + _, err := q.redisCli.Eval(pending2ReadyScript, keys, []interface{}{now}) + if err != nil && err != NilErr { return fmt.Errorf("pending2ReadyScript failed: %v", err) } return nil @@ -235,10 +254,9 @@ return msg func (q *DelayQueue) ready2Unack() (string, error) { retryTime := time.Now().Add(q.maxConsumeDuration).Unix() - ctx := context.Background() keys := []string{q.readyKey, q.unAckKey} - ret, err := q.redisCli.Eval(ctx, ready2UnackScript, keys, retryTime).Result() - if err == redis.Nil { + ret, err := q.redisCli.Eval(ready2UnackScript, keys, []interface{}{retryTime}) + if err == NilErr { return "", err } if err != nil { @@ -253,11 +271,10 @@ func (q *DelayQueue) ready2Unack() (string, error) { func (q *DelayQueue) retry2Unack() (string, error) { retryTime := time.Now().Add(q.maxConsumeDuration).Unix() - ctx := context.Background() keys := []string{q.retryKey, q.unAckKey} - ret, err := q.redisCli.Eval(ctx, ready2UnackScript, keys, retryTime, q.retryKey, q.unAckKey).Result() - if err == redis.Nil { - return "", redis.Nil + ret, err := q.redisCli.Eval(ready2UnackScript, keys, []interface{}{retryTime, q.retryKey, q.unAckKey}) + if err == NilErr { + return "", NilErr } if err != nil { return "", fmt.Errorf("ready2UnackScript failed: %v", err) @@ -270,9 +287,8 @@ func (q *DelayQueue) retry2Unack() (string, error) { } func (q *DelayQueue) callback(idStr string) error { - ctx := context.Background() - payload, err := q.redisCli.Get(ctx, q.genMsgKey(idStr)).Result() - if err == redis.Nil { + payload, err := q.redisCli.Get(q.genMsgKey(idStr)) + if err == NilErr { return nil } if err != nil { @@ -326,24 +342,21 @@ func (q *DelayQueue) batchCallback(ids []string) { } func (q *DelayQueue) ack(idStr string) error { - ctx := context.Background() - err := q.redisCli.ZRem(ctx, q.unAckKey, idStr).Err() + err := q.redisCli.ZRem(q.unAckKey, []string{idStr}) if err != nil { return fmt.Errorf("remove from unack failed: %v", err) } // msg key has ttl, ignore result of delete - _ = q.redisCli.Del(ctx, q.genMsgKey(idStr)).Err() - q.redisCli.HDel(ctx, q.retryCountKey, idStr) + _ = q.redisCli.Del([]string{q.genMsgKey(idStr)}) + _ = q.redisCli.HDel(q.retryCountKey, []string{idStr}) return nil } func (q *DelayQueue) nack(idStr string) error { - ctx := context.Background() // update retry time as now, unack2Retry will move it to retry immediately - err := q.redisCli.ZAdd(ctx, q.unAckKey, redis.Z{ - Member: idStr, - Score: float64(time.Now().Unix()), - }).Err() + err := q.redisCli.ZAdd(q.unAckKey, map[string]float64{ + idStr: float64(time.Now().Unix()), + }) if err != nil { return fmt.Errorf("negative ack failed: %v", err) } @@ -392,19 +405,17 @@ redis.call('ZRemRangeByScore', KEYS[1], '0', ARGV[1]) -- remove msgs from unack ` func (q *DelayQueue) unack2Retry() error { - ctx := context.Background() keys := []string{q.unAckKey, q.retryCountKey, q.retryKey, q.garbageKey} now := time.Now() - err := q.redisCli.Eval(ctx, unack2RetryScript, keys, now.Unix()).Err() - if err != nil && err != redis.Nil { + _, err := q.redisCli.Eval(unack2RetryScript, keys, []interface{}{now.Unix()}) + if err != nil && err != NilErr { return fmt.Errorf("unack to retry script failed: %v", err) } return nil } func (q *DelayQueue) garbageCollect() error { - ctx := context.Background() - msgIds, err := q.redisCli.SMembers(ctx, q.garbageKey).Result() + msgIds, err := q.redisCli.SMembers(q.garbageKey) if err != nil { return fmt.Errorf("smembers failed: %v", err) } @@ -416,12 +427,12 @@ func (q *DelayQueue) garbageCollect() error { for _, idStr := range msgIds { msgKeys = append(msgKeys, q.genMsgKey(idStr)) } - err = q.redisCli.Del(ctx, msgKeys...).Err() - if err != nil && err != redis.Nil { + err = q.redisCli.Del(msgKeys) + if err != nil && err != NilErr { return fmt.Errorf("del msgs failed: %v", err) } - err = q.redisCli.SRem(ctx, q.garbageKey, msgIds).Err() - if err != nil && err != redis.Nil { + err = q.redisCli.SRem(q.garbageKey, msgIds) + if err != nil && err != NilErr { return fmt.Errorf("remove from garbage key failed: %v", err) } return nil @@ -437,7 +448,7 @@ func (q *DelayQueue) consume() error { ids := make([]string, 0, q.fetchLimit) for { idStr, err := q.ready2Unack() - if err == redis.Nil { // consumed all + if err == NilErr { // consumed all break } if err != nil { @@ -464,7 +475,7 @@ func (q *DelayQueue) consume() error { ids = make([]string, 0, q.fetchLimit) for { idStr, err := q.retry2Unack() - if err == redis.Nil { // consumed all + if err == NilErr { // consumed all break } if err != nil { diff --git a/delayqueue_test.go b/delayqueue_test.go index 9c29839..15949b4 100644 --- a/delayqueue_test.go +++ b/delayqueue_test.go @@ -57,6 +57,46 @@ func TestDelayQueue_consume(t *testing.T) { } } +func TestDelayQueueOnCluster(t *testing.T) { + redisCli := redis.NewClusterClient(&redis.ClusterOptions{ + Addrs: []string{ + "127.0.0.1:7000", + "127.0.0.1:7001", + "127.0.0.1:7002", + }, + }) + redisCli.FlushDB(context.Background()) + size := 1000 + succeed := 0 + cb := func(s string) bool { + succeed++ + return true + } + queue := NewQueueOnCluster("test", redisCli, cb). + WithFetchInterval(time.Millisecond * 50). + WithMaxConsumeDuration(0). + WithLogger(log.New(os.Stderr, "[DelayQueue]", log.LstdFlags)). + WithFetchLimit(2). + WithConcurrent(1) + + for i := 0; i < size; i++ { + err := queue.SendDelayMsg(strconv.Itoa(i), 0) + if err != nil { + t.Error(err) + } + } + for i := 0; i < 10*size; i++ { + err := queue.consume() + if err != nil { + t.Errorf("consume error: %v", err) + return + } + } + if succeed != size { + t.Error("msg not consumed") + } +} + func TestDelayQueue_ConcurrentConsume(t *testing.T) { redisCli := redis.NewClient(&redis.Options{ Addr: "127.0.0.1:6379", diff --git a/go.mod b/go.mod index df3c66b..23a47c8 100644 --- a/go.mod +++ b/go.mod @@ -4,5 +4,5 @@ go 1.16 require ( github.com/google/uuid v1.3.0 - github.com/redis/go-redis/v9 v9.0.5 // indirect + github.com/redis/go-redis/v9 v9.0.5 ) diff --git a/go.sum b/go.sum index d107bb3..c01307c 100644 --- a/go.sum +++ b/go.sum @@ -1,105 +1,12 @@ +github.com/bsm/ginkgo/v2 v2.7.0 h1:ItPMPH90RbmZJt5GtkcNvIRuGEdwlBItdNVoyzaNQao= github.com/bsm/ginkgo/v2 v2.7.0/go.mod h1:AiKlXPm7ItEHNc/2+OkrNG4E0ITzojb9/xWzvQ9XZ9w= +github.com/bsm/gomega v1.26.0 h1:LhQm+AFcgV2M0WyKroMASzAzCAJVpAxQXv4SaI9a69Y= github.com/bsm/gomega v1.26.0/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0= -github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE= -github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cespare/xxhash/v2 v2.2.0 h1:DC2CZ1Ep5Y4k3ZQ899DldepgrayRUGE6BBZ/cd9Cj44= github.com/cespare/xxhash/v2 v2.2.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= -github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78= github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc= -github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo= -github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4= -github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ= -github.com/go-redis/redis/v8 v8.11.4 h1:kHoYkfZP6+pe04aFTnhDH6GDROa5yJdHJVNxV3F46Tg= -github.com/go-redis/redis/v8 v8.11.4/go.mod h1:2Z2wHZXdQpCDXEGzqMockDpNyYvi2l4Pxt6RJr792+w= -github.com/go-task/slim-sprig v0.0.0-20210107165309-348f09dbbbc0/go.mod h1:fyg7847qk6SyHyPtNmDHnmrv/HOrqktSC+C9fM+CJOE= -github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U= -github.com/golang/protobuf v1.4.0-rc.1/go.mod h1:ceaxUfeHdC40wWswd/P6IGgMaK3YpKi5j83Wpe3EHw8= -github.com/golang/protobuf v1.4.0-rc.1.0.20200221234624-67d41d38c208/go.mod h1:xKAWHe0F5eneWXFV3EuXVDTCmh+JuBKY0li0aMyXATA= -github.com/golang/protobuf v1.4.0-rc.2/go.mod h1:LlEzMj4AhA7rCAGe4KMBDvJI+AwstrUpVNzEA03Pprs= -github.com/golang/protobuf v1.4.0-rc.4.0.20200313231945-b860323f09d0/go.mod h1:WU3c8KckQ9AFe+yFwt9sWVRKCVIyN9cPHBJSNnbL67w= -github.com/golang/protobuf v1.4.0/go.mod h1:jodUvKwWbYaEsadDk5Fwe5c77LiNKVO9IDvqG2KuDX0= -github.com/golang/protobuf v1.4.2/go.mod h1:oDoupMAO8OvCJWAcko0GGGIgR6R6ocIYbsSw735rRwI= -github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk= -github.com/golang/protobuf v1.5.2/go.mod h1:XVQd3VNwM+JqD3oG2Ue2ip4fOMUkwXdXDdiuN0vRsmY= -github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= -github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= -github.com/google/go-cmp v0.4.0/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= github.com/google/uuid v1.3.0 h1:t6JiXgmwXMjEs8VusXIJk2BXHsn+wx8BZdTaoZ5fu7I= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/hpcloud/tail v1.0.0/go.mod h1:ab1qPbhIpdTxEkNHXyeSf5vhxWSCs/tWer42PpOxQnU= -github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A= -github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE= -github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU= -github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE= -github.com/onsi/ginkgo v1.12.1/go.mod h1:zj2OWP4+oCPe1qIXoGWkgMRwljMUYCdkwsT2108oapk= -github.com/onsi/ginkgo v1.16.4 h1:29JGrr5oVBm5ulCWet69zQkzWipVXIol6ygQUe/EzNc= -github.com/onsi/ginkgo v1.16.4/go.mod h1:dX+/inL/fNMqNlz0e9LfyB9TswhZpCVdJM/Z6Vvnwo0= -github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7JYyY= -github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo= -github.com/onsi/gomega v1.16.0 h1:6gjqkI8iiRHMvdccRJM8rVKjCWk6ZIm6FTm3ddIe4/c= -github.com/onsi/gomega v1.16.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY= -github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/redis/go-redis/v9 v9.0.5 h1:CuQcn5HIEeK7BgElubPP8CGtE0KakrnbBSTLjathl5o= github.com/redis/go-redis/v9 v9.0.5/go.mod h1:WqMKv5vnQbRuZstUwxQI195wHy+t4PuXDOjzMvcuQHk= -github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= -github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA= -github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= -golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= -golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= -golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= -golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= -golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4= -golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= -golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= -golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A= -golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.0.0-20210428140749-89ef3d95e781 h1:DzZ89McO9/gWPsQXS/FVKAlG02ZjaQ6AlZRBimEYOd0= -golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk= -golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= -golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20190904154756-749cb33beabd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20191005200804-aed5e4c7ecf9/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20191120155948-bd437916bb0e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200323222414-85ca7c5b95cd/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210112080510-489259a85091/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.0.0-20210423082822-04245dca01da h1:b3NXsE2LusjYGGjL5bxEVZZORm/YEFFrWFjR8eFrw/c= -golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= -golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= -golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M= -golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= -golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= -golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8= -google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0= -google.golang.org/protobuf v0.0.0-20200228230310-ab0ca4ff8a60/go.mod h1:cfTl7dwQJ+fmap5saPgwCLgHXTUD7jkjRqWcaiX5VyM= -google.golang.org/protobuf v1.20.1-0.20200309200217-e05f789c0967/go.mod h1:A+miEFZTKqfCUM6K7xSMQL9OKL/b6hQv+e19PK+JZNE= -google.golang.org/protobuf v1.21.0/go.mod h1:47Nbq4nVaFHyn7ilMalzfO3qCViNmqZ2kzikPIcrTAo= -google.golang.org/protobuf v1.23.0/go.mod h1:EGpADcykh3NcUnDUJcl1+ZksZNG86OlYog2l/sGQquU= -google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= -google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys= -gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7 h1:uRGJdciOHaEIrze2W8Q3AKkepLTh2hOroT7a+7czfdQ= -gopkg.in/tomb.v1 v1.0.0-20141024135613-dd632973f1e7/go.mod h1:dt/ZhP58zS4L8KSrWDmTeBkI65Dw0HsyUHuEVlX15mw= -gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.2.4/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.3.0/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI= -gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY= -gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ= diff --git a/wrapper.go b/wrapper.go new file mode 100644 index 0000000..6c28f69 --- /dev/null +++ b/wrapper.go @@ -0,0 +1,173 @@ +package delayqueue + +import ( + "context" + "github.com/redis/go-redis/v9" + "time" +) + +func NewQueue(name string, cli *redis.Client, callback func(string) bool, opts ...interface{}) *DelayQueue { + rc := &redisV9Wrapper{ + inner: cli, + } + return NewQueue0(name, rc, callback, opts...) +} + +func wrapErr(err error) error { + if err == redis.Nil { + return NilErr + } + return err +} + +type redisV9Wrapper struct { + inner *redis.Client +} + +func (r *redisV9Wrapper) Eval(script string, keys []string, args []interface{}) (interface{}, error) { + ctx := context.Background() + ret, err := r.inner.Eval(ctx, script, keys, args...).Result() + return ret, wrapErr(err) +} + +func (r *redisV9Wrapper) Set(key string, value string, expiration time.Duration) error { + ctx := context.Background() + return wrapErr(r.inner.Set(ctx, key, value, expiration).Err()) +} + +func (r *redisV9Wrapper) Get(key string) (string, error) { + ctx := context.Background() + ret, err := r.inner.Get(ctx, key).Result() + return ret, wrapErr(err) +} + +func (r *redisV9Wrapper) Del(keys []string) error { + ctx := context.Background() + return wrapErr(r.inner.Del(ctx, keys...).Err()) +} + +func (r *redisV9Wrapper) HSet(key string, field string, value string) error { + ctx := context.Background() + return wrapErr(r.inner.HSet(ctx, key, field, value).Err()) +} + +func (r *redisV9Wrapper) HDel(key string, fields []string) error { + ctx := context.Background() + return wrapErr(r.inner.HDel(ctx, key, fields...).Err()) +} + +func (r *redisV9Wrapper) SMembers(key string) ([]string, error) { + ctx := context.Background() + ret, err := r.inner.SMembers(ctx, key).Result() + return ret, wrapErr(err) +} + +func (r *redisV9Wrapper) SRem(key string, members []string) error { + ctx := context.Background() + members2 := make([]interface{}, len(members)) + for i, v := range members { + members2[i] = v + } + return wrapErr(r.inner.SRem(ctx, key, members2...).Err()) +} + +func (r *redisV9Wrapper) ZAdd(key string, values map[string]float64) error { + ctx := context.Background() + var zs []redis.Z + for member, score := range values { + zs = append(zs, redis.Z{ + Score: score, + Member: member, + }) + } + return wrapErr(r.inner.ZAdd(ctx, key, zs...).Err()) +} + +func (r *redisV9Wrapper) ZRem(key string, members []string) error { + ctx := context.Background() + members2 := make([]interface{}, len(members)) + for i, v := range members { + members2[i] = v + } + return wrapErr(r.inner.ZRem(ctx, key, members2...).Err()) +} + +type redisClusterWrapper struct { + inner *redis.ClusterClient +} + +func (r *redisClusterWrapper) Eval(script string, keys []string, args []interface{}) (interface{}, error) { + ctx := context.Background() + ret, err := r.inner.Eval(ctx, script, keys, args...).Result() + return ret, wrapErr(err) +} + +func (r *redisClusterWrapper) Set(key string, value string, expiration time.Duration) error { + ctx := context.Background() + return wrapErr(r.inner.Set(ctx, key, value, expiration).Err()) +} + +func (r *redisClusterWrapper) Get(key string) (string, error) { + ctx := context.Background() + ret, err := r.inner.Get(ctx, key).Result() + return ret, wrapErr(err) +} + +func (r *redisClusterWrapper) Del(keys []string) error { + ctx := context.Background() + return wrapErr(r.inner.Del(ctx, keys...).Err()) +} + +func (r *redisClusterWrapper) HSet(key string, field string, value string) error { + ctx := context.Background() + return wrapErr(r.inner.HSet(ctx, key, field, value).Err()) +} + +func (r *redisClusterWrapper) HDel(key string, fields []string) error { + ctx := context.Background() + return wrapErr(r.inner.HDel(ctx, key, fields...).Err()) +} + +func (r *redisClusterWrapper) SMembers(key string) ([]string, error) { + ctx := context.Background() + ret, err := r.inner.SMembers(ctx, key).Result() + return ret, wrapErr(err) +} + +func (r *redisClusterWrapper) SRem(key string, members []string) error { + ctx := context.Background() + members2 := make([]interface{}, len(members)) + for i, v := range members { + members2[i] = v + } + return wrapErr(r.inner.SRem(ctx, key, members2...).Err()) +} + +func (r *redisClusterWrapper) ZAdd(key string, values map[string]float64) error { + ctx := context.Background() + var zs []redis.Z + for member, score := range values { + zs = append(zs, redis.Z{ + Score: score, + Member: member, + }) + } + return wrapErr(r.inner.ZAdd(ctx, key, zs...).Err()) +} + +func (r *redisClusterWrapper) ZRem(key string, members []string) error { + ctx := context.Background() + members2 := make([]interface{}, len(members)) + for i, v := range members { + members2[i] = v + } + return wrapErr(r.inner.ZRem(ctx, key, members2...).Err()) +} + +func NewQueueOnCluster(name string, cli *redis.ClusterClient, callback func(string) bool, opts ...interface{}) *DelayQueue { + rc := &redisClusterWrapper{ + inner: cli, + } + opts = append(opts, UseHashTagKey()) + return NewQueue0(name, rc, callback, opts...) +}