调整Key以支持Redis集群版

Merge remote-tracking branch 'refs/remotes/origin/dev' into dev
优化添加的逻辑
2026-05-26 21:03:50 +08:00 · 2026-05-17 21:48:38 +08:00 · 2025-10-13 19:01:45 +08:00
10 changed files with 336 additions and 35 deletions
@@ -81,12 +81,12 @@ 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 有序集合
+		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(),
@@ -462,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)
@@ -483,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)
 		}
@@ -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)
+			}
+		})
+	}
+}
@@ -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
+}
@@ -170,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)
@@ -187,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,
@@ -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,
@@ -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,
@@ -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,10 +114,10 @@ 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,
+		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,
@@ -411,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,
@@ -431,20 +439,20 @@ func (l *Once) Create(ctx context.Context, taskType OnceTaskType, taskId string,
 		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
@@ -452,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) {
@@ -471,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)
@@ -566,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
@@ -617,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)
 }
@@ -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()
 }
@@ -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,
@@ -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,
 		}
Author	SHA1	Message	Date
Yun	fa359b2b1a	调整Key以支持Redis集群版	2026-05-26 21:03:50 +08:00
yun	f1e8a3e965	Merge remote-tracking branch 'refs/remotes/origin/dev' into dev	2026-05-17 21:48:38 +08:00
yun	382bc7b21b	优化添加的逻辑	2025-10-13 19:01:45 +08:00