调整Key以支持Redis集群版

cluster.go

@@ -81,13 +81,13 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 		logger:         op.logger,
 		keyPrefix:      keyPrefix,
 		location:       op.location,
-		lockKey:        "timer:cluster_globalLockKey" + keyPrefix,  // 定时器的全局锁
-		zsetKey:        "timer:cluster_zsetKey" + keyPrefix,        // 有序集合
-		listKey:        "timer:cluster_listKey" + keyPrefix,        // 列表
-		setKey:         "timer:cluster_setKey" + keyPrefix,         // 重入集合
-		priorityKey:    "timer:cluster_priorityKey" + keyPrefix,    // 全局优先级的key
-		executeInfoKey: "timer:cluster_executeInfoKey" + keyPrefix, // 执行情况的key 有序集合
-		usePriority:    op.usePriority,
+		lockKey:        fmt.Sprintf("timer:{%s}:cluster_lock", keyPrefix),
+		zsetKey:        fmt.Sprintf("timer:{%s}:cluster_zset", keyPrefix),
+		listKey:        fmt.Sprintf("timer:{%s}:cluster_list", keyPrefix),
+		setKey:         fmt.Sprintf("timer:{%s}:cluster_set", keyPrefix),
+		priorityKey:    fmt.Sprintf("timer:{%s}:cluster_priority", keyPrefix),
+		executeInfoKey: fmt.Sprintf("timer:{%s}:cluster_execinfo", keyPrefix),
+		usePriority:    false,
 		stopChan:       make(chan struct{}),
 		instanceId:     U.String(),
 		cronParser:     op.cronParser,
@@ -98,7 +98,18 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 
 	// 初始化优先级
 
-	if clu.usePriority {
+	if op.priorityType != priorityTypeNone {
+
+		clu.usePriority = true
+
+		if op.priorityType == priorityTypeVersion {
+			pVal, err := priority.PriorityByVersion(op.priorityVersion)
+			if err != nil {
+				clu.logger.Errorf(ctx, "PriorityByVersion version:%s err:%v", op.priorityVersion, err)
+				return nil, err
+			}
+			op.priorityVal = pVal
+		}
 
 		pri, err := priority.InitPriority(
 			ctx,
@@ -341,15 +352,14 @@ func (c *Cluster) EveryMinute(ctx context.Context, taskId string, second int, ca
 
 // 特定时间间隔
 func (c *Cluster) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) error {
-	nowTime := time.Now().In(c.location)
 
 	if spaceTime < 0 {
 		c.logger.Errorf(ctx, "间隔时间不能小于0")
 		return errors.New("间隔时间不能小于0")
 	}
 
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location)
 
 	jobData := JobData{
 		JobType:      JobTypeInterval,
@@ -370,9 +380,8 @@ func (c *Cluster) EverySpace(ctx context.Context, taskId string, spaceTime time.
 // @param extendData interface{} 扩展数据
 // @return error
 func (l *Cluster) Cron(ctx context.Context, taskId string, cronExpression string, callback func(ctx context.Context, extendData any) error, extendData any, opt ...Option) error {
-	nowTime := time.Now().In(l.location)
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, l.location)
 
 	options := newEmptyOptions(opt...)
 	cronParser := l.cronParser
@@ -453,10 +462,10 @@ func (l *Cluster) calculateNextTimes() {
 		// 使用Lua脚本原子性添加任务
 		script := `
 			local zsetKey = KEYS[1]
+			local lockKey = KEYS[2]
 			local score = ARGV[1]
 			local taskID = ARGV[2]
 			local expireTime = ARGV[3]
-			local lockKey = ARGV[4]
 
 			-- 检查是否已存在
 			local existing = redis.call('zscore', zsetKey, taskID)
@@ -474,9 +483,9 @@ func (l *Cluster) calculateNextTimes() {
 			return 1
 			`
 
-		lockKey := fmt.Sprintf("%s_%s_%d", l.keyPrefix, val.TaskId, nextTime.UnixMilli())
-		_, err = pipe.Eval(l.ctx, script, []string{l.zsetKey},
-			nextTime.UnixMilli(), val.TaskId, 60, lockKey).Result()
+		lockKey := fmt.Sprintf("timer:{%s}:cluster_calc_lock:%s:%d", l.keyPrefix, val.TaskId, nextTime.UnixMilli())
+		_, err = pipe.Eval(l.ctx, script, []string{l.zsetKey, lockKey},
+			nextTime.UnixMilli(), val.TaskId, 60).Result()
 		if err != nil {
 			l.logger.Errorf(l.ctx, "Failed to schedule task: %v", err)
 		}
@@ -530,28 +539,34 @@ func (c *Cluster) executeTasks() {
 		case <-c.ctx.Done():
 			return
 		case c.workerChan <- struct{}{}:
-			if c.usePriority && !c.priority.IsLatest(c.ctx) {
-				time.Sleep(5 * time.Second)
-				continue
-			}
+			func() {
+				defer func() {
+					<-c.workerChan
+				}()
 
-			taskID, err := c.redis.BLPop(c.ctx, 10*time.Second, c.listKey).Result()
-			if err != nil {
-				if err != redis.Nil {
-					c.logger.Errorf(c.ctx, "Failed to pop task: %v", err)
-					// Redis 异常，休眠一会儿
+				if c.usePriority && !c.priority.IsLatest(c.ctx) {
 					time.Sleep(5 * time.Second)
+					return
 				}
-				continue
-			}
 
-			if len(taskID) < 2 {
-				c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID)
-				// 数据异常，继续下一个
-				continue
-			}
+				taskID, err := c.redis.BLPop(c.ctx, 10*time.Second, c.listKey).Result()
+				if err != nil {
+					if err != redis.Nil {
+						c.logger.Errorf(c.ctx, "Failed to pop task: %v", err)
+						// Redis 异常，休眠一会儿
+						time.Sleep(5 * time.Second)
+					}
+					return
+				}
 
-			go c.processTask(taskID[1])
+				if len(taskID) < 2 {
+					c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID)
+					// 数据异常，继续下一个
+					return
+				}
+
+				go c.processTask(taskID[1])
+			}()
 		}
 	}
 
@@ -564,9 +579,6 @@ type ReJobData struct {
 
 // 执行任务
 func (l *Cluster) processTask(taskId string) {
-	defer func() {
-		<-l.workerChan
-	}()
 
 	begin := time.Now()
 

cluster_test.go

@@ -3,6 +3,7 @@ package timerx_test
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 	"time"
 
@@ -182,3 +183,80 @@ func TestCluster_Add(t *testing.T) {
 
 	// TODO: verify the job is added to the cluster and can be executed after the specified duration
 }
+
+func TestClusterKeyFormat(t *testing.T) {
+	keyPrefix := "testcluster"
+
+	tests := []struct {
+		name      string
+		key       string
+		hashTag   string
+		hasPrefix string
+	}{
+		{"zsetKey", "timer:{testcluster}:cluster_zset", "{testcluster}", "timer:"},
+		{"listKey", "timer:{testcluster}:cluster_list", "{testcluster}", "timer:"},
+		{"lockKey", "timer:{testcluster}:cluster_lock", "{testcluster}", "timer:"},
+		{"execInfoKey", "timer:{testcluster}:cluster_execinfo", "{testcluster}", "timer:"},
+		{"priorityKey", "timer:{testcluster}:cluster_priority", "{testcluster}", "timer:"},
+		{"calcLockKey", "timer:{testcluster}:cluster_calc_lock:task1:1234567890", "{testcluster}", "timer:"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if tt.key == "" || tt.key == " " {
+				t.Error("key should not be empty")
+			}
+			t.Logf("key format: %s", tt.key)
+		})
+	}
+
+	_ = keyPrefix
+}
+
+func TestClusterZsetListKeysSameHashTag(t *testing.T) {
+	zsetKey := "timer:{myapp}:cluster_zset"
+	listKey := "timer:{myapp}:cluster_list"
+
+	zsetTag := extractHashTag(zsetKey)
+	listTag := extractHashTag(listKey)
+
+	if zsetTag != listTag {
+		t.Errorf("zset and list keys must share same hash tag: %q != %q", zsetTag, listTag)
+	}
+	if zsetTag != "myapp" {
+		t.Errorf("unexpected hash tag: %q", zsetTag)
+	}
+}
+
+func extractHashTag(key string) string {
+	start := strings.Index(key, "{")
+	end := strings.Index(key, "}")
+	if start >= 0 && end > start {
+		return key[start+1 : end]
+	}
+	return ""
+}
+
+func TestClusterMultiKeyLuaKeysShareSlot(t *testing.T) {
+	type keyPair struct {
+		key1 string
+		key2 string
+		name string
+	}
+
+	pairs := []keyPair{
+		{"timer:{app}:cluster_zset", "timer:{app}:cluster_list", "zset+list"},
+		{"timer:{app}:cluster_zset", "timer:{app}:cluster_calc_lock:task1:9999", "zset+calc_lock"},
+	}
+
+	for _, p := range pairs {
+		t.Run(p.name, func(t *testing.T) {
+			t1 := extractHashTag(p.key1)
+			t2 := extractHashTag(p.key2)
+			if t1 != t2 {
+				t.Errorf("keys in multi-key Lua script must share same hash tag: %q != %q (%s, %s)",
+					t1, t2, p.key1, p.key2)
+			}
+		})
+	}
+}

example/cluster/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -19,7 +20,7 @@ func main() {
 
 	ops := []timerx.Option{}
 
-	clu, err := timerx.InitCluster(ctx, client, "cluster", ops...)
+	clu, err := timerx.InitCluster(ctx, client, "cluster_01", ops...)
 	if err != nil {
 		panic(err)
 	}
@@ -39,22 +40,43 @@ func main() {
 // space
 func space(ctx context.Context, clu *timerx.Cluster) {
 	// 每秒执行一次
-	err := clu.EverySpace(ctx, "space_test_second", 1*time.Second, callback, "space 这是秒任务")
+	err := clu.EverySpace(ctx, "space_test_second_1", 1*time.Second, callback, "space 这是秒任务")
 	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_second_5", 5*time.Second, callback, "space 这是5秒任务")
+	fmt.Println(err)
+
 	// 每分钟执行一次
-	err = clu.EverySpace(ctx, "space_test_minute", 1*time.Minute, callback, "space 这是分钟任务")
+	err = clu.EverySpace(ctx, "space_test_minute_1", 1*time.Minute, callback, "space 这是分钟任务")
 	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_minute_5", 5*time.Minute, callback, "space 这是5分钟任务")
+	fmt.Println(err)
+
 	// 每小时执行一次
-	err = clu.EverySpace(ctx, "space_test_hour", 1*time.Hour, callback, "space 这是小时任务")
+	err = clu.EverySpace(ctx, "space_test_hour_1", 1*time.Hour, callback, "space 这是小时任务")
 	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_2", 2*time.Hour, callback, "space 这是2小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_3", 3*time.Hour, callback, "space 这是3小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_4", 4*time.Hour, callback, "space 这是4小时任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_hour_5", 5*time.Hour, callback, "space 这是5小时任务")
+	fmt.Println(err)
+
 	// 每天执行一次
-	err = clu.EverySpace(ctx, "space_test_day", 24*time.Hour, callback, "space 这是天任务")
+	err = clu.EverySpace(ctx, "space_test_day_1", 24*time.Hour, callback, "space 这是天任务")
 	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_day_2", 2*24*time.Hour, callback, "space 这是2天任务")
+	fmt.Println(err)
+	err = clu.EverySpace(ctx, "space_test_day_3", 3*24*time.Hour, callback, "space 这是3天任务")
+	fmt.Println(err)
+
 	// 每周执行一次
-	err = clu.EverySpace(ctx, "space_test_week", 7*24*time.Hour, callback, "space 这是周任务")
+	err = clu.EverySpace(ctx, "space_test_week_1", 7*24*time.Hour, callback, "space 这是7天任务")
 	fmt.Println(err)
+
 	// 每月执行一次
-	err = clu.EverySpace(ctx, "space_test_month", 30*24*time.Hour, callback, "space 这是月任务")
+	err = clu.EverySpace(ctx, "space_test_month_1", 30*24*time.Hour, callback, "space 这是30天任务")
 	fmt.Println(err)
 
 }
@@ -252,6 +274,13 @@ func getRedis() *redis.Client {
 func callback(ctx context.Context, extendData any) error {
 
 	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path,"/cache/cluster/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
 
 	// 追加到文件
 	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)

example/cluster_redis/main.go

@@ -0,0 +1,57 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+	"github.com/yuninks/timerx"
+)
+
+func main() {
+	ctx := context.Background()
+
+	client := getClusterRedis()
+
+	clu, err := timerx.InitCluster(ctx, client, "cluster_redis_example")
+	if err != nil {
+		panic(err)
+	}
+	defer clu.Stop()
+
+	err = clu.EverySpace(ctx, "cluster_every_second", 1*time.Second, clusterCallback, "cluster每秒任务")
+	fmt.Println("EverySpace:", err)
+
+	err = clu.EveryMinute(ctx, "cluster_every_minute", 30, clusterCallback, "cluster每分钟任务")
+	fmt.Println("EveryMinute:", err)
+
+	err = clu.EveryDay(ctx, "cluster_every_day", 0, 0, 0, clusterCallback, "cluster每天任务")
+	fmt.Println("EveryDay:", err)
+
+	err = clu.Cron(ctx, "cluster_cron_second", "*/5 * * * * ?", clusterCallback, "cluster cron任务",
+		timerx.WithCronParserSecond())
+	fmt.Println("Cron:", err)
+
+	select {}
+}
+
+func getClusterRedis() *redis.ClusterClient {
+	client := redis.NewClusterClient(&redis.ClusterOptions{
+		Addrs: []string{
+			"127.0.0.1:6379",
+			"127.0.0.1:6380",
+			"127.0.0.1:6381",
+		},
+		Password: "",
+	})
+	if client == nil {
+		panic("redis cluster init error")
+	}
+	return client
+}
+
+func clusterCallback(ctx context.Context, extendData any) error {
+	fmt.Println("cluster任务执行:", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
+	return nil
+}

example/once/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -22,7 +23,7 @@ func main() {
 	ctx := context.Background()
 
 	client := getRedis()
-	once, err := timerx.InitOnce(ctx, client, "once_test", &OnceWorker{}, timerx.WithBatchSize(1000))
+	once, err := timerx.InitOnce(ctx, client, "once_01", &OnceWorker{}, timerx.WithBatchSize(1000))
 	if err != nil {
 		panic(err)
 	}
@@ -30,13 +31,74 @@ func main() {
 	// intervalSaveTime(ctx, once)
 	// intervalSave(ctx, once)
 	// intervalcreateTask(ctx, once)
-	intervalCreateTime(ctx, once)
+	// intervalCreateTime(ctx, once)
 	// benchmarkJob(ctx, once)
+	stabilityTest(ctx, once)
 
 	select {}
 
 }
 
+// 稳定性测试
+func stabilityTest(ctx context.Context, once *timerx.Once) {
+
+	timer := time.NewTicker(time.Second)
+
+	for {
+		select {
+		case t := <-timer.C:
+			fmt.Println("time:", t)
+
+			str := t.Format("2006-01-02 15:04:05")
+
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_0_%d", time.Now().Unix()), 0, "Create 间隔0s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_1_%d", time.Now().Unix()), time.Second*1, "Create 间隔1s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_10_%d", time.Now().Unix()), time.Second*10, "Create 间隔10s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_15_%d", time.Now().Unix()), time.Second*15, "Create 间隔15s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_30_%d", time.Now().Unix()), time.Second*30, "Create 间隔30s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_60_%d", time.Now().Unix()), time.Second*60, "Create 间隔60s ["+str+"]")
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_120_%d", time.Now().Unix()), time.Second*120, "Create 间隔120s ["+str+"]")    // 2分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_300_%d", time.Now().Unix()), time.Second*300, "Create 间隔300s ["+str+"]")    // 5分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_900_%d", time.Now().Unix()), time.Second*900, "Create 间隔900s ["+str+"]")    // 15分钟
+			once.Create(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_1800_%d", time.Now().Unix()), time.Second*1800, "Create 间隔1800s ["+str+"]") // 30分钟
+
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_0_%d", time.Now().Unix()), 0, "Save 间隔0s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_1_%d", time.Now().Unix()), time.Second*1, "Save 间隔1s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_10_%d", time.Now().Unix()), time.Second*10, "Save 间隔10s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_15_%d", time.Now().Unix()), time.Second*15, "Save 间隔15s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_30_%d", time.Now().Unix()), time.Second*30, "Save 间隔30s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_60_%d", time.Now().Unix()), time.Second*60, "Save 间隔60s ["+str+"]")
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_120_%d", time.Now().Unix()), time.Second*120, "Save 间隔120s ["+str+"]")    // 2分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_300_%d", time.Now().Unix()), time.Second*300, "Save 间隔300s ["+str+"]")    // 5分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_900_%d", time.Now().Unix()), time.Second*900, "Save 间隔900s ["+str+"]")    // 15分钟
+			once.Save(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_1800_%d", time.Now().Unix()), time.Second*1800, "Save 间隔1800s ["+str+"]") // 30分钟
+
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_0_%d", time.Now().Unix()), time.Now(), "CreateByTime 当前时间 ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_1_%d", time.Now().Unix()), time.Now().Add(time.Second*1), "CreateByTime 当前时间+1s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_10_%d", time.Now().Unix()), time.Now().Add(time.Second*10), "CreateByTime 当前时间+10s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_15_%d", time.Now().Unix()), time.Now().Add(time.Second*15), "CreateByTime 当前时间+15s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_30_%d", time.Now().Unix()), time.Now().Add(time.Second*30), "CreateByTime 当前时间+30s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_60_%d", time.Now().Unix()), time.Now().Add(time.Second*60), "CreateByTime 当前时间+60s ["+str+"]")
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_120_%d", time.Now().Unix()), time.Now().Add(time.Second*120), "CreateByTime 当前时间+120s ["+str+"]")    // 2分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_300_%d", time.Now().Unix()), time.Now().Add(time.Second*300), "CreateByTime 当前时间+300s ["+str+"]")    // 5分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_900_%d", time.Now().Unix()), time.Now().Add(time.Second*900), "CreateByTime 当前时间+900s ["+str+"]")    // 15分钟
+			once.CreateByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_create_by_1800_%d", time.Now().Unix()), time.Now().Add(time.Second*1800), "CreateByTime 当前时间+1800s ["+str+"]") // 30分钟
+
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_0_%d", time.Now().Unix()), time.Now(), "SaveByTime 当前时间 ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_1_%d", time.Now().Unix()), time.Now().Add(time.Second*1), "SaveByTime 当前时间+1s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_10_%d", time.Now().Unix()), time.Now().Add(time.Second*10), "SaveByTime 当前时间+10s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_15_%d", time.Now().Unix()), time.Now().Add(time.Second*15), "SaveByTime 当前时间+15s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_30_%d", time.Now().Unix()), time.Now().Add(time.Second*30), "SaveByTime 当前时间+30s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_60_%d", time.Now().Unix()), time.Now().Add(time.Second*60), "SaveByTime 当前时间+60s ["+str+"]")
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_120_%d", time.Now().Unix()), time.Now().Add(time.Second*120), "SaveByTime 当前时间+120s ["+str+"]")    // 2分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_300_%d", time.Now().Unix()), time.Now().Add(time.Second*300), "SaveByTime 当前时间+300s ["+str+"]")    // 5分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_900_%d", time.Now().Unix()), time.Now().Add(time.Second*900), "SaveByTime 当前时间+900s ["+str+"]")    // 15分钟
+			once.SaveByTime(ctx, OnceTaskTypeNormal, fmt.Sprintf("stabilityTest_task_save_by_1800_%d", time.Now().Unix()), time.Now().Add(time.Second*1800), "SaveByTime 当前时间+1800s ["+str+"]") // 30分钟
+		}
+	}
+
+}
+
 // 指定时间测试
 func intervalSaveTime(ctx context.Context, once *timerx.Once) {
 
@@ -108,7 +170,7 @@ func benchmarkJob(ctx context.Context, once *timerx.Once) {
 	ch := make(chan ChanStatus, 1000)
 
 	go func() {
-		for a := 0; a < 100; a++ {
+		for a := 0; a < 10; a++ {
 			go func(a int) {
 				for status := range ch {
 					// fmt.Println("协程", a, "处理任务", status)
@@ -125,9 +187,9 @@ func benchmarkJob(ctx context.Context, once *timerx.Once) {
 	go func() {
 		// 一千万任务，每个任务间隔1秒
 
-		for i := 0; i < 100; i++ {
+		for i := 0; i < 10; i++ {
 			runTime := t.Add(time.Duration(i) * time.Second)
-			for j := 0; j < 100; j++ {
+			for j := 0; j < 10; j++ {
 				ch <- ChanStatus{
 					I: i,
 					J: j,
@@ -170,6 +232,14 @@ func callback(ctx context.Context, extendData any) error {
 
 	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
 
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path, "/cache/once/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
+
 	// 追加到文件
 	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
 	if err != nil {
@@ -177,7 +247,7 @@ func callback(ctx context.Context, extendData any) error {
 		return err
 	}
 	defer file.Close()
-	_, err = file.WriteString(fmt.Sprintf("执行时间:%v %s\n", extendData, time.Now().Format("2006-01-02 15:04:05")))
+	_, err = file.WriteString(fmt.Sprintf("执行时间:%v [%s] \n", extendData, time.Now().Format("2006-01-02 15:04:05")))
 	if err != nil {
 		fmt.Println("写入文件失败:", err)
 		return err

example/single/main.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"os"
+	"path/filepath"
 	"time"
 
 	"github.com/redis/go-redis/v9"
@@ -35,22 +36,44 @@ func main() {
 // space
 func space(ctx context.Context, clu *timerx.Single) {
 	// 每秒执行一次
-	_, err := clu.EverySpace(ctx, "space_test_second", 1*time.Second, callback, "space 这是秒任务")
+	_, err := clu.EverySpace(ctx, "space_test_second_1", 1*time.Second, callback, "space 这是秒任务")
 	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_second_5", 5*time.Second, callback, "space 这是5秒任务")
+	fmt.Println(err)
+
 	// 每分钟执行一次
-	_, err = clu.EverySpace(ctx, "space_test_minute", 1*time.Minute, callback, "space 这是分钟任务")
+	_, err = clu.EverySpace(ctx, "space_test_minute_1", 1*time.Minute, callback, "space 这是分钟任务")
 	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_minute_5", 5*time.Minute, callback, "space 这是5分钟任务")
+	fmt.Println(err)
+
 	// 每小时执行一次
-	_, err = clu.EverySpace(ctx, "space_test_hour", 1*time.Hour, callback, "space 这是小时任务")
+	_, err = clu.EverySpace(ctx, "space_test_hour_1", 1*time.Hour, callback, "space 这是小时任务")
 	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_2", 2*time.Hour, callback, "space 这是2小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_3", 3*time.Hour, callback, "space 这是3小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_4", 4*time.Hour, callback, "space 这是4小时任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_hour_5", 5*time.Hour, callback, "space 这是5小时任务")
+	fmt.Println(err)
+
 	// 每天执行一次
-	_, err = clu.EverySpace(ctx, "space_test_day", 24*time.Hour, callback, "space 这是天任务")
+	_, err = clu.EverySpace(ctx, "space_test_day_1", 24*time.Hour, callback, "space 这是天任务")
 	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_day_2", 2*24*time.Hour, callback, "space 这是2天任务")
+	fmt.Println(err)
+	_, err = clu.EverySpace(ctx, "space_test_day_3", 3*24*time.Hour, callback, "space 这是3天任务")
+	fmt.Println(err)
+
 	// 每周执行一次
-	_, err = clu.EverySpace(ctx, "space_test_week", 7*24*time.Hour, callback, "space 这是周任务")
+	_, err = clu.EverySpace(ctx, "space_test_week_1", 7*24*time.Hour, callback, "space 这是周任务")
 	fmt.Println(err)
+
+
 	// 每月执行一次
-	_, err = clu.EverySpace(ctx, "space_test_month", 30*24*time.Hour, callback, "space 这是月任务")
+	_, err = clu.EverySpace(ctx, "space_test_month_1", 30*24*time.Hour, callback, "space 这是月任务")
 	fmt.Println(err)
 
 }
@@ -249,6 +272,14 @@ func callback(ctx context.Context, extendData any) error {
 
 	fmt.Println("任务执行了", extendData, "时间:", time.Now().Format("2006-01-02 15:04:05"))
 
+	// 解析文件路径，每天一个文件
+	path, _ := filepath.Abs("./")
+	// 拼接文件路径
+	save_path = filepath.Join(path, "/cache/single/"+time.Now().Format("2006-01-02")+".log")
+	// 创建文件夹
+	dir := filepath.Dir(save_path)
+	os.MkdirAll(dir, 0755)
+
 	// 追加到文件
 	file, err := os.OpenFile(save_path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
 	if err != nil {

go.mod

@@ -7,8 +7,8 @@ require (
 	github.com/redis/go-redis/v9 v9.14.0
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/stretchr/testify v1.11.1
-	github.com/yuninks/cachex v1.0.5
-	github.com/yuninks/lockx v1.1.3
+	github.com/yuninks/cachex v1.0.6
+	github.com/yuninks/lockx v1.1.4
 )
 
 require (

go.sum

@@ -25,10 +25,10 @@ github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY=
 github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
-github.com/yuninks/cachex v1.0.5 h1:Y2NmTsuEgwEVYb7FVFh5tUN67kmrUioeksQqLbOAwsM=
-github.com/yuninks/cachex v1.0.5/go.mod h1:5357qz18UvHTJSgZzkMamUzZoFzGeKG9+4tIUBXRSVM=
-github.com/yuninks/lockx v1.1.3 h1:OA6rb4/XOj+M+1vKLs8fqsU4ZCadvga+oARTWTHpwx0=
-github.com/yuninks/lockx v1.1.3/go.mod h1:+HyRozwQHMHrykyOFlotV4Z+z2yrgRSdDl8TxxRMFzw=
+github.com/yuninks/cachex v1.0.6 h1:G5dJ8O0gLAGnLwydEHVCSSGv+p1z8KfZdjb5NdxxsVA=
+github.com/yuninks/cachex v1.0.6/go.mod h1:5357qz18UvHTJSgZzkMamUzZoFzGeKG9+4tIUBXRSVM=
+github.com/yuninks/lockx v1.1.4 h1:Wrd4aAU5apNWAamZM7yqoNlwHDjGY/X1YYrLXcgj1+s=
+github.com/yuninks/lockx v1.1.4/go.mod h1:+HyRozwQHMHrykyOFlotV4Z+z2yrgRSdDl8TxxRMFzw=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=

heartbeat/heartbeat.go

@@ -3,6 +3,7 @@ package heartbeat
 import (
 	"context"
 	"errors"
+	"fmt"
 	"strconv"
 	"sync"
 	"time"
@@ -47,7 +48,7 @@ func InitHeartBeat(ctx context.Context, ref redis.UniversalClient, keyPrefix str
 		ctx:    ctx,
 		cancel: cancel,
 
-		heartbeatKey: "timer:heartbeat_key" + op.source + keyPrefix,
+		heartbeatKey: fmt.Sprintf("timer:{%s}:heartbeat_%s", keyPrefix, op.source),
 
 		priority:   op.priority,
 		redis:      ref,

leader/leader.go

@@ -5,6 +5,7 @@ package leader
 import (
 	"context"
 	"errors"
+	"fmt"
 	"sync"
 	"time"
 
@@ -52,8 +53,8 @@ func InitLeader(ctx context.Context, ref redis.UniversalClient, keyPrefix string
 		ctx:              ctx,
 		cancel:           cancel,
 		redis:            ref,
-		leaderUniLockKey: "timer:leader_lockKey" + op.source + keyPrefix,
-		leaderKey:        "timer:leader" + op.source + keyPrefix,
+		leaderUniLockKey: fmt.Sprintf("timer:{%s}:leader_lock_%s", keyPrefix, op.source),
+		leaderKey:        fmt.Sprintf("timer:{%s}:leader_%s", keyPrefix, op.source),
 		priority:         op.priority,
 		instanceId:       op.instanceId,
 		logger:           op.logger,

once.go

@@ -90,6 +90,13 @@ const (
 	jobTypeList = "list"
 )
 
+type createSource string
+
+const (
+	createSourceDefault = "default"
+	createSourceRetry   = "retry"
+)
+
 // 初始化
 func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, call Callback, opts ...Option) (*Once, error) {
 	op := newOptions(opts...)
@@ -107,11 +114,11 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 		ctx:              ctx,
 		cancel:           cancel,
 		logger:           op.logger,
-		zsetKey:          "timer:once_zsetkey" + keyPrefix,
-		listKey:          "timer:once_listkey" + keyPrefix,
-		executeInfoKey:   "timer:once_executeInfoKey" + keyPrefix,
-		globalLockPrefix: "timer:once_globalLockPrefix" + keyPrefix,
-		usePriority:      op.usePriority,
+		zsetKey:          fmt.Sprintf("timer:{%s}:once_zset", keyPrefix),
+		listKey:          fmt.Sprintf("timer:{%s}:once_list", keyPrefix),
+		executeInfoKey:   fmt.Sprintf("timer:{%s}:once_execinfo", keyPrefix),
+		globalLockPrefix: fmt.Sprintf("timer:{%s}:once_lock:", keyPrefix),
+		usePriority:      false,
 		redis:            re,
 		worker:           call,
 		keyPrefix:        keyPrefix,
@@ -126,7 +133,19 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 	}
 
 	// 初始化优先级
-	if wo.usePriority {
+	if op.priorityType != priorityTypeNone {
+
+		wo.usePriority = true
+
+		if op.priorityType == priorityTypeVersion {
+			pVal, err := priority.PriorityByVersion(op.priorityVersion)
+			if err != nil {
+				wo.logger.Errorf(ctx, "PriorityByVersion version:%s err:%v", op.priorityVersion, err)
+				return nil, err
+			}
+			op.priorityVal = pVal
+		}
+
 		pri, err := priority.InitPriority(
 			ctx,
 			re,
@@ -277,38 +296,40 @@ func (l *Once) executeTasks() {
 
 	for {
 
-		if l.usePriority {
-			if !l.priority.IsLatest(l.ctx) {
-				time.Sleep(time.Second * 5)
-				continue
-			}
-		}
-
 		select {
 		case <-l.stopChan:
 			return
 		case <-l.ctx.Done():
 			return
 		case l.workerChan <- struct{}{}:
+			func() {
+				defer func() {
+					<-l.workerChan
+				}()
 
-			keys, err := l.redis.BLPop(l.ctx, time.Second*10, l.listKey).Result()
-			if err != nil {
-				if err != redis.Nil {
-					l.logger.Errorf(l.ctx, "Failed to pop task: %v", err)
-					// Redis 异常，休眠一会儿再重试
+				if l.usePriority && !l.priority.IsLatest(l.ctx) {
 					time.Sleep(time.Second * 5)
+					return
 				}
-				continue
-			}
 
-			if len(keys) < 2 {
-				l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
-				// 数据异常，继续下一个
-				continue
-			}
-			// 处理任务
-			go l.processTask(keys[1])
+				keys, err := l.redis.BLPop(l.ctx, time.Second*10, l.listKey).Result()
+				if err != nil {
+					if err != redis.Nil {
+						l.logger.Errorf(l.ctx, "Failed to pop task: %v", err)
+						// Redis 异常，休眠一会儿再重试
+						time.Sleep(time.Second * 5)
+					}
+					return
+				}
 
+				if len(keys) < 2 {
+					l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
+					// 数据异常，继续下一个
+					return
+				}
+				// 处理任务
+				go l.processTask(keys[1])
+			}()
 		}
 	}
 
@@ -335,6 +356,9 @@ func (l *Once) parseRedisKey(key string) (OnceTaskType, string, error) {
 // @param delayTime time.Duration 延迟时间
 // @param attachData interface{} 附加数据
 func (l *Once) Save(ctx context.Context, taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any) error {
+	if delayTime < 0 {
+		delayTime = 0
+	}
 	execTime := time.Now().Add(delayTime)
 	return l.save(ctx, jobTypeOnce, taskType, taskId, []time.Time{execTime}, attachData, 0)
 }
@@ -394,7 +418,8 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType,
 
 	dataExpire := time.Until(expiresTime)
 
-	pipe.SetEx(w.ctx, w.keyPrefix+redisKey, b, dataExpire)
+	dataKey := fmt.Sprintf("timer:{%s}:once_data:%s", w.keyPrefix, redisKey)
+	pipe.SetEx(w.ctx, dataKey, b, dataExpire)
 	pipe.ZAdd(w.ctx, w.zsetKey, redis.Z{
 		Score:  float64(nextTime.UnixMilli()),
 		Member: redisKey,
@@ -411,24 +436,23 @@ func (w *Once) save(ctx context.Context, jobType jobType, taskType OnceTaskType,
 // 添加任务(不覆盖)
 func (l *Once) Create(ctx context.Context, taskType OnceTaskType, taskId string, delayTime time.Duration, attachData any) error {
 	if delayTime <= 0 {
-		l.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v", taskType, taskId, attachData)
-		return ErrDelayTime
+		delayTime = 0
 	}
 	execTime := time.Now().Add(delayTime)
-	return l.create(ctx, jobTypeOnce, taskType, taskId, []time.Time{execTime}, attachData, 0)
+	return l.create(ctx, createSourceDefault, jobTypeOnce, taskType, taskId, []time.Time{execTime}, attachData, 0)
 }
 
 // 指定时间执行(不覆盖)
 func (l *Once) CreateByTime(ctx context.Context, taskType OnceTaskType, taskId string, executeTime time.Time, attachData any) error {
-	return l.create(ctx, jobTypeOnce, taskType, taskId, []time.Time{executeTime}, attachData, 0)
+	return l.create(ctx, createSourceDefault, jobTypeOnce, taskType, taskId, []time.Time{executeTime}, attachData, 0)
 }
 
 // 指定多个时间执行(不覆盖)
 func (l *Once) CreateByList(ctx context.Context, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any) error {
-	return l.create(ctx, jobTypeList, taskType, taskId, taskTimes, attachData, 0)
+	return l.create(ctx, createSourceDefault, jobTypeList, taskType, taskId, taskTimes, attachData, 0)
 }
 
-func (l *Once) create(ctx context.Context, jobType jobType, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any, runCount int) error {
+func (l *Once) create(ctx context.Context, source createSource, jobType jobType, taskType OnceTaskType, taskId string, taskTimes []time.Time, attachData any, runCount int) error {
 	if len(taskTimes) <= 0 {
 		l.logger.Errorf(ctx, "delay time must be positive taskType:%v taskId:%v attachData:%v runCount:%v", taskType, taskId, attachData, runCount)
 		return ErrExecuteTime
@@ -436,6 +460,20 @@ func (l *Once) create(ctx context.Context, jobType jobType, taskType OnceTaskTyp
 
 	redisKey := l.buildRedisKey(taskType, taskId)
 
+	if source != createSourceRetry {
+		// 这里加一个全局锁 从Retry来的不需要 因为已经加锁了
+		lock, err := lockx.NewGlobalLock(ctx, l.redis, l.globalLockPrefix+redisKey)
+		if err != nil {
+			l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s", taskId)
+			return err
+		}
+		if b, err := lock.Lock(); !b {
+			l.logger.Errorf(ctx, "processTask timer:获取锁失败:%s %+v", taskId, err)
+			return err
+		}
+		defer lock.Unlock()
+	}
+
 	score, err := l.redis.ZScore(l.ctx, l.zsetKey, redisKey).Result()
 	if err != nil {
 		if errors.Is(err, redis.Nil) {
@@ -455,9 +493,10 @@ func (l *Once) create(ctx context.Context, jobType jobType, taskType OnceTaskTyp
 // 删除任务
 func (w *Once) Delete(taskType OnceTaskType, taskId string) error {
 	redisKey := w.buildRedisKey(taskType, taskId)
+	dataKey := fmt.Sprintf("timer:{%s}:once_data:%s", w.keyPrefix, redisKey)
 
 	pipe := w.redis.TxPipeline()
-	pipe.Del(w.ctx, redisKey)
+	pipe.Del(w.ctx, dataKey)
 	pipe.ZRem(w.ctx, w.zsetKey, redisKey)
 
 	_, err := pipe.Exec(w.ctx)
@@ -506,9 +545,6 @@ func (l *Once) batchGetTasks() {
 
 // 执行任务
 func (l *Once) processTask(key string) {
-	defer func() {
-		<-l.workerChan
-	}()
 
 	begin := time.Now()
 
@@ -553,8 +589,8 @@ func (l *Once) processTask(key string) {
 	}()
 
 	// 读取数据
-	redisKey := l.keyPrefix + l.buildRedisKey(taskType, taskId)
-	str, err := l.redis.Get(ctx, redisKey).Result()
+	dataKey := fmt.Sprintf("timer:{%s}:once_data:%s", l.keyPrefix, l.buildRedisKey(taskType, taskId))
+	str, err := l.redis.Get(ctx, dataKey).Result()
 	if err != nil {
 		l.logger.Errorf(ctx, "processTask redis.Get key:%s err:%s", key, err)
 		return
@@ -604,5 +640,5 @@ func (l *Once) handleRetry(ctx context.Context, taskType OnceTaskType, taskId st
 		taskType, taskId, ed.RunCount)
 
 	// 不覆盖的新建
-	return l.create(ctx, ed.JobType, taskType, taskId, ed.TaskTimes, ed.Data, ed.RunCount)
+	return l.create(ctx, createSourceRetry, ed.JobType, taskType, taskId, ed.TaskTimes, ed.Data, ed.RunCount)
 }

once_test.go

@@ -1,7 +1,148 @@
 package timerx
 
-import "testing"
+import (
+	"strings"
+	"testing"
+	"time"
 
-func Test2(t *testing.T) {
+	"github.com/yuninks/timerx/logger"
+)
 
+func TestOnceBuildRedisKey(t *testing.T) {
+	once := &Once{keySeparator: "[:]", keyPrefix: "test_prefix"}
+
+	key := once.buildRedisKey("normal", "task123")
+	if key != "normal[:]task123" {
+		t.Errorf("unexpected key: %s", key)
+	}
+}
+
+func TestOnceParseRedisKey(t *testing.T) {
+	once := &Once{keySeparator: "[:]"}
+
+	taskType, taskId, err := once.parseRedisKey("normal[:]task123")
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if taskType != "normal" {
+		t.Errorf("unexpected taskType: %s", taskType)
+	}
+	if taskId != "task123" {
+		t.Errorf("unexpected taskId: %s", taskId)
+	}
+
+	_, _, err = once.parseRedisKey("invalid")
+	if err == nil {
+		t.Error("expected error for invalid key")
+	}
+}
+
+func TestOnceKeyFormat(t *testing.T) {
+	keyPrefix := "testcluster"
+
+	tests := []struct {
+		name     string
+		key      string
+		contains string
+	}{
+		{"zsetKey contains hash tag", "timer:{testcluster}:once_zset", "{testcluster}"},
+		{"listKey contains hash tag", "timer:{testcluster}:once_list", "{testcluster}"},
+		{"execinfoKey contains hash tag", "timer:{testcluster}:once_execinfo", "{testcluster}"},
+		{"lockPrefix contains hash tag", "timer:{testcluster}:once_lock:", "{testcluster}"},
+		{"dataKey contains hash tag", "timer:{testcluster}:once_data:normal[:]task1", "{testcluster}"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if !strings.Contains(tt.key, tt.contains) {
+				t.Errorf("key %q does not contain hash tag %q", tt.key, tt.contains)
+			}
+		})
+	}
+
+	_ = keyPrefix
+}
+
+func TestOnceDataKeyFormat(t *testing.T) {
+	dataKey := "timer:{myapp}:once_data:normal[:]task1"
+
+	if !strings.Contains(dataKey, "{myapp}") {
+		t.Errorf("data key %q missing hash tag {myapp}", dataKey)
+	}
+	if !strings.HasPrefix(dataKey, "timer:") {
+		t.Errorf("data key %q missing timer: prefix", dataKey)
+	}
+}
+
+func TestOnceExtendData(t *testing.T) {
+	taskTimes := []time.Time{
+		time.Now().Add(10 * time.Second),
+		time.Now().Add(5 * time.Second),
+		time.Now().Add(15 * time.Second),
+	}
+
+	ed := extendData{
+		TaskTimes: taskTimes,
+		Data:      "test data",
+		RunCount:  0,
+		JobType:   jobTypeOnce,
+	}
+
+	if len(ed.TaskTimes) != 3 {
+		t.Errorf("expected 3 task times, got %d", len(ed.TaskTimes))
+	}
+	if ed.Data != "test data" {
+		t.Errorf("unexpected data: %v", ed.Data)
+	}
+	if ed.RunCount != 0 {
+		t.Errorf("unexpected runCount: %d", ed.RunCount)
+	}
+}
+
+func TestOnceSaveValidation(t *testing.T) {
+	var once Once
+
+	once = Once{keyPrefix: "test_app", keySeparator: "[:]"}
+
+	if once.keySeparator != "[:]" {
+		t.Errorf("unexpected keySeparator: %s", once.keySeparator)
+	}
+	if once.keyPrefix != "test_app" {
+		t.Errorf("unexpected keyPrefix: %s", once.keyPrefix)
+	}
+}
+
+func TestOnceKeyPrefixConsistency(t *testing.T) {
+	keyPrefix := "myapp_v2"
+
+	zsetKey := "timer:{myapp_v2}:once_zset"
+	listKey := "timer:{myapp_v2}:once_list"
+	dataKey := "timer:{myapp_v2}:once_data:urgent[:]task99"
+
+	for _, k := range []string{zsetKey, listKey, dataKey} {
+		if !strings.Contains(k, "{"+keyPrefix+"}") {
+			t.Errorf("key %q missing hash tag for keyPrefix %q", k, keyPrefix)
+		}
+	}
+
+	zsetHash := extractHashTagForOnce(zsetKey)
+	listHash := extractHashTagForOnce(listKey)
+	dataHash := extractHashTagForOnce(dataKey)
+
+	if zsetHash != listHash || listHash != dataHash {
+		t.Errorf("all Once keys must share same hash tag: %q %q %q", zsetHash, listHash, dataHash)
+	}
+}
+
+func extractHashTagForOnce(key string) string {
+	start := strings.Index(key, "{")
+	end := strings.Index(key, "}")
+	if start >= 0 && end > start {
+		return key[start+1 : end]
+	}
+	return ""
+}
+
+func newDefaultLoggerForTest() logger.Logger {
+	return logger.NewLogger()
 }

option.go

@@ -8,32 +8,41 @@ import (
 )
 
 type Options struct {
-	logger      logger.Logger
-	location    *time.Location
-	timeout     time.Duration // 任务最长执行时间
-	usePriority bool
-	priorityVal int64
-	batchSize   int
-	maxRunCount int          // 单个任务最大运行次数 0代表不限
-	maxWorkers  int          // 最大工作协程数
-	cronParser  *cron.Parser // cron表达式解析器
+	logger          logger.Logger
+	location        *time.Location
+	timeout         time.Duration // 任务最长执行时间
+	priorityType    priorityType  // 策略类型 0.不使用 1.优先级 2.版本
+	priorityVal     int64         // 策略优先级
+	priorityVersion string        // 策略版本的集
+	batchSize       int
+	maxRunCount     int          // 单个任务最大运行次数 0代表不限
+	maxWorkers      int          // 最大工作协程数
+	cronParser      *cron.Parser // cron表达式解析器
 }
 
+type priorityType int8
+
+const (
+	priorityTypeNone     priorityType = 0 // 不使用优先级
+	priorityTypePriority priorityType = 1
+	priorityTypeVersion  priorityType = 2 // 版本
+)
+
 func defaultOptions() Options {
 
 	// 默认使用Linux的定时任务兼容
 	parser := cron.NewParser(cron.Minute | cron.Hour | cron.Dom | cron.Month | cron.Dow)
 
 	return Options{
-		logger:      logger.NewLogger(),
-		location:    time.Local,
-		timeout:     time.Hour, //
-		usePriority: false,
-		priorityVal: 0,
-		batchSize:   100,
-		maxRunCount: 0,
-		maxWorkers:  100,
-		cronParser:  &parser,
+		logger:       logger.NewLogger(),
+		location:     time.Local,
+		timeout:      time.Hour, //
+		priorityType: priorityTypeNone,
+		priorityVal:  0,
+		batchSize:    100,
+		maxRunCount:  0,
+		maxWorkers:   100,
+		cronParser:   &parser,
 	}
 }
 
@@ -81,11 +90,18 @@ func WithTimeout(d time.Duration) Option {
 // 设置优先级
 func WithPriority(priority int64) Option {
 	return func(o *Options) {
-		o.usePriority = true
+		o.priorityType = priorityTypePriority
 		o.priorityVal = priority
 	}
 }
 
+func WithPriorityByVersion(version string) Option {
+	return func(o *Options) {
+		o.priorityType = priorityTypeVersion
+		o.priorityVersion = version
+	}
+}
+
 func WithBatchSize(size int) Option {
 	return func(o *Options) {
 		if size <= 1 {

priority/priority.go

@@ -50,7 +50,7 @@ func InitPriority(ctx context.Context, re redis.UniversalClient, keyPrefix strin
 		priority:    priority,
 		redis:       re,
 		logger:      conf.logger,
-		redisKey:    "timer:priority_" + conf.source + keyPrefix,
+		redisKey:    fmt.Sprintf("timer:{%s}:priority_%s", keyPrefix, conf.source),
 		expireTime:  conf.expireTime,
 		setInterval: conf.updateInterval,
 		getInterval: conf.getInterval,

priority/priority_test.go

@@ -131,8 +131,8 @@ func TestSetPriorityScenarios(t *testing.T) {
 	ctx := context.Background()
 
 	redisConn := getRedis()
-	// 删除Key
-	redisConn.Del(ctx, "timer:priority_test22")
+	// 删除Key (new key format with hash tag)
+	redisConn.Del(ctx, "timer:{test22}:priority_")
 
 	for _, tc := range testCases {
 		t.Run(tc.name, func(t *testing.T) {
@@ -197,7 +197,7 @@ func TestErrorScenarios(t *testing.T) {
 
 		priority := &Priority{
 			redis:    getRedis(),
-			redisKey: "timer:priority_test",
+			redisKey: "timer:{test}:priority_",
 			priority: 100,
 			ctx:      ctx,
 		}

priority/version_test.go

@@ -6,7 +6,6 @@ import (
 	"github.com/yuninks/timerx/priority"
 )
 
-
 func TestVersionToPriority(t *testing.T) {
 	tests := []struct {
 		name    string
@@ -20,6 +19,18 @@ func TestVersionToPriority(t *testing.T) {
 			want:    1002003000000,
 			wantErr: false,
 		},
+		{
+			name:    "standard version0",
+			version: "0.0.0",
+			want:    0,
+			wantErr: false,
+		},
+				{
+			name:    "standard version1",
+			version: "1.0.0",
+			want:    1000000000000,
+			wantErr: false,
+		},
 		{
 			name:    "version with v prefix",
 			version: "v1.2.3",

readme.md

@@ -14,6 +14,7 @@
 
 1. 针对月的任务，需要注意日期有效性，且在月末的最后一天，需要考虑月末的最后一天的下一个任务执行时间
 2. 集群部署时，存在新旧的代码混合问题，任务调度可能存在问题（需要根据实际需要进行版本上线/下线操作）
+3. 主从切换时也要做到平滑上下线
 
 ## 方案一
 

single.go

@@ -252,15 +252,14 @@ func (c *Single) EveryMinute(ctx context.Context, taskId string, second int, cal
 
 // 特定时间间隔
 func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.Duration, callback func(ctx context.Context, extendData interface{}) error, extendData interface{}) (int64, error) {
-	nowTime := time.Now().In(c.location)
 
 	if spaceTime < 0 {
 		c.logger.Errorf(ctx, "间隔时间不能小于0")
 		return 0, ErrIntervalTime
 	}
 
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, c.location)
 
 	jobData := JobData{
 		JobType: JobTypeInterval,
@@ -274,9 +273,8 @@ func (c *Single) EverySpace(ctx context.Context, taskId string, spaceTime time.D
 }
 
 func (l *Single) Cron(ctx context.Context, taskId string, cronExpression string, callback func(ctx context.Context, extendData any) error, extendData any, opt ...Option) (int64, error) {
-	nowTime := time.Now().In(l.location)
-	// 获取当天的零点时间
-	zeroTime := time.Date(nowTime.Year(), nowTime.Month(), nowTime.Day(), 0, 0, 0, 0, nowTime.Location())
+	// 固定时间点为20250101 00:00:00，便于计算下一次执行时间
+	zeroTime := time.Date(2025, 1, 1, 0, 0, 0, 0, l.location)
 
 	options := Options{}
 	for _, o := range opt {