修复一个单次的BUG

cmd/main.go

@@ -24,12 +24,68 @@ func main() {
 
 	// re()
 	// d()
-	cluster()
+	// cluster()
+	once()
 
 	select {}
 
 }
 
+func once() {
+	client := getRedis()
+	ctx := context.Background()
+	w := OnceWorker{}
+	one := timerx.InitOnce(ctx, client, "test", w)
+
+	d := OnceData{
+		Num: 3,
+	}
+	// dy, _ := json.Marshal(d)
+
+	err := one.Create("test", "test3", 1*time.Second, d)
+	if err != nil {
+		fmt.Println(err)
+	}
+	d = OnceData{
+		Num: 4,
+	}
+	// dy, _ = json.Marshal(d)
+	err = one.Create("test", "test4", 1*time.Second, d)
+	if err != nil {
+		fmt.Println(err)
+	}
+
+}
+
+type OnceData struct {
+	Num int
+}
+
+type OnceWorker struct{}
+
+func (l OnceWorker) Worker(ctx context.Context, taskType string, taskId string, attachData interface{}) *timerx.OnceWorkerResp {
+	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
+	fmt.Println(taskType, taskId)
+
+	fmt.Printf("原来的参数：%+v\n", attachData)
+
+	// d := OnceData{}
+
+	// json.Unmarshal(ab, &d)
+
+	// d.Num++
+
+	// fmt.Println(d)
+
+	// dy, _ := json.Marshal(d)
+
+	return &timerx.OnceWorkerResp{
+		Retry:      true,
+		AttachData: attachData,
+		DelayTime:  1 * time.Second,
+	}
+}
+
 func cluster() {
 	client := getRedis()
 	ctx := context.Background()
@@ -45,36 +101,27 @@ func cluster() {
 }
 
 func worker() {
-	client := getRedis()
-	w := timerx.InitOnce(context.Background(), client, "test", &Worker{})
-	w.Add("test", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test2", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test3", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test4", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
-	w.Add("test5", "test", 1*time.Second, map[string]interface{}{
-		"test": "test",
-	})
+	// client := getRedis()
+	// w := timerx.InitOnce(context.Background(), client, "test", &OnceWorker{})
+	// w.Save("test", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test2", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test3", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test4", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
+	// w.Save("test5", "test", 1*time.Second, map[string]interface{}{
+	// 	"test": "test",
+	// })
 
 	select {}
 }
 
-type Worker struct{}
-
-func (w *Worker) Worker(ctx context.Context, jobType string, uniqueKey string, data interface{}) (timerx.WorkerCode, time.Duration) {
-	fmt.Println("执行时间:", time.Now().Format("2006-01-02 15:04:05"))
-	fmt.Println(uniqueKey, jobType)
-	fmt.Println(data)
-	return timerx.WorkerCodeAgain, time.Second
-}
-
 func getRedis() *redis.Client {
 	client := redis.NewClient(&redis.Options{
 		Addr:     "127.0.0.1" + ":" + "6379",

once.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"log"
 	"runtime/debug"
 	"strings"
 	"sync"
@@ -21,20 +20,20 @@ import (
 
 // 单次的任务队列
 type Once struct {
-	ctx     context.Context
-	logger  Logger
-	zsetKey string
-	listKey string
-	redis   redis.UniversalClient
-	worker  Callback
+	ctx       context.Context
+	logger    Logger
+	zsetKey   string
+	listKey   string
+	redis     redis.UniversalClient
+	worker    Callback
+	keyPrefix string
 }
 
-type WorkerCode int
-
-const (
-	WorkerCodeSuccess WorkerCode = 0  // 处理完成(不需要重入)
-	WorkerCodeAgain   WorkerCode = -1 // 需要继续定时,默认原来的时间
-)
+type OnceWorkerResp struct {
+	Retry      bool // 是否重试 true
+	DelayTime  time.Duration
+	AttachData interface{}
+}
 
 // 需要考虑执行失败重新放入队列的情况
 type Callback interface {
@@ -44,7 +43,7 @@ type Callback interface {
 	// @param data interface{} 任务数据
 	// @return WorkerCode 任务执行结果
 	// @return time.Duration 任务执行时间间隔
-	Worker(ctx context.Context, jobType string, uniTaskId string, attachData interface{}) (WorkerCode, time.Duration)
+	Worker(ctx context.Context, taskType string, taskId string, attachData interface{}) *OnceWorkerResp
 }
 
 var wo *Once = nil
@@ -56,16 +55,21 @@ type extendData struct {
 }
 
 // 初始化
-func InitOnce(ctx context.Context, re *redis.Client, keyPrefix string, call Callback, opts ...Option) *Once {
+func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, call Callback, opts ...Option) *Once {
+	if re == nil {
+		panic("redis client is nil")
+	}
+
 	op := newOptions(opts...)
 	once.Do(func() {
 		wo = &Once{
-			ctx:     ctx,
-			logger:  op.logger,
-			zsetKey: "timer:once_zsetkey" + keyPrefix,
-			listKey: "timer:once_listkey" + keyPrefix,
-			redis:   re,
-			worker:  call,
+			ctx:       ctx,
+			logger:    op.logger,
+			zsetKey:   "timer:once_zsetkey" + keyPrefix,
+			listKey:   "timer:once_listkey" + keyPrefix,
+			redis:     re,
+			worker:    call,
+			keyPrefix: keyPrefix,
 		}
 		go wo.getTask()
 		go wo.watch()
@@ -74,13 +78,13 @@ func InitOnce(ctx context.Context, re *redis.Client, keyPrefix string, call Call
 	return wo
 }
 
-// 添加任务
+// 添加任务(覆盖)
 // 重复插入就代表覆盖
 // @param jobType string 任务类型
 // @param uniTaskId string 任务唯一标识
 // @param delayTime time.Duration 延迟时间
 // @param attachData interface{} 附加数据
-func (w *Once) Add(jobType string, uniTaskId string, delayTime time.Duration, attachData interface{}) error {
+func (w *Once) Save(taskType string, taskId string, delayTime time.Duration, attachData interface{}) error {
 	if delayTime.Abs() != delayTime {
 		return fmt.Errorf("时间间隔不能为负数")
 	}
@@ -88,7 +92,7 @@ func (w *Once) Add(jobType string, uniTaskId string, delayTime time.Duration, at
 		return fmt.Errorf("时间间隔不能为0")
 	}
 
-	redisKey := fmt.Sprintf("%s[:]%s", jobType, uniTaskId)
+	redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
 
 	ed := extendData{
 		Delay: delayTime,
@@ -97,12 +101,12 @@ func (w *Once) Add(jobType string, uniTaskId string, delayTime time.Duration, at
 	b, _ := json.Marshal(ed)
 
 	// 写入附加数据
-	_, err := w.redis.SetEX(w.ctx, redisKey, b, delayTime+time.Second*5).Result()
+	_, err := w.redis.SetEX(w.ctx, w.keyPrefix+redisKey, b, delayTime+time.Second*5).Result()
 	if err != nil {
 		return err
 	}
 
-	// 吸入执行时间
+	// 写入执行时间
 	_, err = w.redis.ZAdd(w.ctx, w.zsetKey, &redis.Z{
 		Score:  float64(time.Now().Add(delayTime).UnixMilli()),
 		Member: redisKey,
@@ -111,9 +115,38 @@ func (w *Once) Add(jobType string, uniTaskId string, delayTime time.Duration, at
 	return err
 }
 
+// 添加任务(不覆盖)
+func (l *Once) Create(taskType string, taskId string, delayTime time.Duration, attachData interface{}) error {
+	redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
+	// 判断有序集合Key是否存在，存在则报错，不存在则写入
+	if l.redis.ZScore(l.ctx, l.zsetKey, redisKey).Val() == 0 {
+
+		ed := extendData{
+			Delay: delayTime,
+			Data:  attachData,
+		}
+		b, _ := json.Marshal(ed)
+
+		// 写入附加数据
+		_, err := l.redis.SetEX(l.ctx, l.keyPrefix+redisKey, b, delayTime+time.Second*5).Result()
+		if err != nil {
+			return err
+		}
+		_, err = l.redis.ZAdd(l.ctx, l.zsetKey, &redis.Z{
+			Score:  float64(time.Now().Add(delayTime).UnixMilli()),
+			Member: redisKey,
+		}).Result()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
 // 删除任务
-func (w *Once) Del(jobType string, uniTaskId string) error {
-	redisKey := fmt.Sprintf("%s[:]%s", jobType, uniTaskId)
+func (w *Once) Delete(taskType string, taskId string) error {
+	redisKey := fmt.Sprintf("%s[:]%s", taskType, taskId)
 
 	w.redis.Del(w.ctx, redisKey).Result()
 
@@ -122,6 +155,11 @@ func (w *Once) Del(jobType string, uniTaskId string) error {
 	return nil
 }
 
+// 获取任务
+func (l *Once) Get(taskType string, taskId string) {
+	//
+}
+
 // 获取任务
 func (w *Once) getTask() {
 	timer := time.NewTicker(time.Millisecond * 200)
@@ -153,47 +191,54 @@ func (w *Once) watch() {
 	for {
 		keys, err := w.redis.BLPop(w.ctx, time.Second*10, w.listKey).Result()
 		if err != nil {
-			fmt.Println("watch err:", err)
+			// fmt.Println("watch err:", err)
 			continue
 		}
 
-		go w.doTask(keys[1])
+		ctx := context.WithValue(w.ctx, "trace_id", uuid.NewV4().String())
+
+		go w.doTask(ctx, keys[1])
 
 	}
 }
 
 // 执行任务
-func (w *Once) doTask(key string) {
+func (l *Once) doTask(ctx context.Context, key string) {
+	fmt.Println("任务时间:", time.Now().Format("2006-01-02 15:04:05"))
 	defer func() {
 		if err := recover(); err != nil {
-			fmt.Println("timer:回调任务panic", err)
-			log.Println("errStack", string(debug.Stack()))
+			l.logger.Errorf(ctx, "timer:回调任务panic:%s stack:%s", err, string(debug.Stack()))
 		}
 	}()
 
 	s := strings.Split(key, "[:]")
 
 	// 读取数据
-	str, err := w.redis.Get(w.ctx, key).Result()
+	redisKey := l.keyPrefix + key
+	str, err := l.redis.Get(ctx, redisKey).Result()
 	if err != nil {
-		fmt.Println("execJob err:", err)
+		l.logger.Errorf(ctx, "获取数据失败 err:%s", err)
 		return
 	}
+
+	fmt.Println("参数：", str)
+
 	ed := extendData{}
 	json.Unmarshal([]byte(str), &ed)
 
-	fmt.Println("开始时间：", time.Now().Format("2006-01-02 15:04:05"))
+	resp := l.worker.Worker(ctx, s[0], s[1], ed.Data)
+	if resp == nil {
+		return
+	}
 
-	ctx := context.WithValue(context.Background(), "trace_id", uuid.NewV4().String)
-
-	code, t := w.worker.Worker(ctx, s[0], s[1], ed.Data)
-
-	if code == WorkerCodeAgain {
+	if resp.Retry {
 		// 重新放入队列
-		fmt.Println("重入时间：", time.Now().Format("2006-01-02 15:04:05"))
-		if t != 0 && t == t.Abs() {
-			ed.Delay = t
+		if resp.DelayTime != 0 && resp.DelayTime == resp.DelayTime.Abs() {
+			ed.Delay = resp.DelayTime
 		}
-		w.Add(s[0], s[1], ed.Delay, ed.Data)
+		ed.Data = resp.AttachData
+		l.logger.Infof(ctx, "任务重新放入队列:%s", key)
+		fmt.Println("重入时间:", time.Now().Format("2006-01-02 15:04:05"))
+		l.Create(s[0], s[1], ed.Delay, ed.Data)
 	}
 }