完成v1版本redis时间队列

redis_pub_sub.go

@@ -14,6 +14,7 @@ import (
 	"git.zhangdeman.cn/zhangdeman/consts"
 	"git.zhangdeman.cn/zhangdeman/event/abstract"
 	"git.zhangdeman.cn/zhangdeman/event/define"
+	"git.zhangdeman.cn/zhangdeman/serialize"
 	"git.zhangdeman.cn/zhangdeman/wrapper"
 	"github.com/redis/go-redis/v9"
 	"time"
@@ -34,6 +35,7 @@ func InitRedisPubSubEvent(redisClient *redis.Client, pubSubConfig *define.RedisE
 		pubSubConfig: pubSubConfig,
 	}
 	instance.SetRedisClient(redisClient, pubSubConfig)
+	instance.StartConsumer() // 启动消费者
 	RedisEventPubSubClient = instance
 }
 
@@ -152,6 +154,25 @@ func (r *RedisEventPubSub) ConsumeEvent(handler abstract.EventHandler, successCa
 	if nil == successCallback {
 		successCallback = define.DefaultSuccessCallbackHandler
 	}
+
+	for eventData := range messageChan {
+		handlerResult, handleErr := handler(eventData)
+		if nil != handleErr {
+			failureCallback(eventData, handlerResult, err)
+		} else {
+			successCallback(eventData, handlerResult)
+		}
+	}
+
+	return nil
+}
+
+// StartConsumer 启动消费者
+//
+// Author : go_developer@163.com<白茶清欢>
+//
+// Date : 18:36 2024/6/27
+func (r *RedisEventPubSub) StartConsumer() {
 	for partition := 0; partition < r.pubSubConfig.PartitionNum; partition++ {
 		go func(realPartition int) {
 			defer func() {
@@ -159,23 +180,23 @@ func (r *RedisEventPubSub) ConsumeEvent(handler abstract.EventHandler, successCa
 					r.panicCallback(panicErr, nil, nil)
 				}
 			}()
-			for !r.isStop || (r.isStop && len(messageChan) == 0) {
-				select {
-				case eventData := <-messageChan:
-					handlerResult, handlerErr := handler(eventData)
-					if nil != handlerErr {
-						// 失败回调
-						failureCallback(eventData, handlerResult, handlerErr)
-					} else {
-						// 成功回调
-						successCallback(eventData, handlerResult)
-					}
+			// 启动消费者
+			realTopic := fmt.Sprintf("%v_%v", r.pubSubConfig.Topic, realPartition)
+			messageChannel := r.redisClient.Subscribe(context.Background(), realTopic).Channel()
+			for message := range messageChannel {
+				var (
+					eventData define.EventData
+					err       error
+				)
+
+				if err = serialize.JSON.UnmarshalWithNumber([]byte(message.Payload), &eventData); nil != err {
+					// TODO : 事件数据解析失败的回调
+					continue
 				}
+				r.messageChan <- &eventData
 			}
 		}(partition)
 	}
-
-	return nil
 }
 
 // Destroy 销毁事件实例