增加redis延迟队列第一版生产者

delay/redis_produce.go

@@ -8,6 +8,11 @@
 package delay
 
 import (
+	"context"
+	"encoding/json"
+	"sync"
+	"time"
+
 	"github.com/go-redis/redis/v8"
 )
 
@@ -34,6 +39,35 @@ type redisProduce struct {
 // Author : go_developer@163.com<白茶清欢>
 //
 // Date : 18:03 2022/7/6
-func (rp *redisProduce) Produce(data *Queue) error {
+func (rp *redisProduce) Produce(ctx context.Context, data ...*Queue) error {
+	if len(data) == 0 {
+		return nil
+	}
+	if nil == ctx {
+		ctx = context.Background()
+	}
+
+	wg := &sync.WaitGroup{}
+	wg.Add(len(data))
+	for _, queueData := range data {
+		go func(inputQueueData *Queue) {
+			defer wg.Done()
+			inputQueueData.err = rp.client.ZAdd(ctx, inputQueueData.Name, rp.buildAddMember(inputQueueData)).Err()
+		}(queueData)
+	}
+	wg.Wait()
 	return nil
 }
+
+// buildAddMember ...
+//
+// Author : go_developer@163.com<白茶清欢>
+//
+// Date : 18:22 2022/7/6
+func (rp *redisProduce) buildAddMember(queueData *Queue) *redis.Z {
+	byteData, _ := json.Marshal(queueData)
+	return &redis.Z{
+		Score:  float64(time.Now().Unix() + queueData.DelayTime),
+		Member: string(byteData),
+	}
+}