From ed326b192db0c5cd387e4ef54f87e9917614516d Mon Sep 17 00:00:00 2001
From: Yun <zm@qq.com>
Date: Wed, 27 May 2026 14:09:08 +0800
Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E9=83=A8=E5=88=86=E5=B9=B6?=
 =?UTF-8?q?=E5=8F=91=E9=97=AE=E9=A2=98?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 cluster.go | 17 +++++++++++------
 keys.go    | 11 +++++++++++
 once.go    | 14 +++++++++-----
 3 files changed, 31 insertions(+), 11 deletions(-)
 create mode 100644 keys.go

diff --git a/cluster.go b/cluster.go
index 09cb949..5f933c1 100644
--- a/cluster.go
+++ b/cluster.go
@@ -70,6 +70,9 @@ func InitCluster(ctx context.Context, red redis.UniversalClient, keyPrefix strin
 
 	ctx, cancel := context.WithCancel(ctx)
 
+	// 清理 keyPrefix 中的 {} 字符，避免破坏 Redis Cluster hash tag
+	keyPrefix = sanitizeKeyPrefix(keyPrefix)
+
 	U, err := uuid.NewV7()
 	if err != nil {
 		op.logger.Errorf(ctx, "InitCluster uuid.NewV7 err:%v, using fallback", err)
@@ -548,10 +551,7 @@ func (c *Cluster) executeTasks() {
 		case <-c.ctx.Done():
 			return
 		case c.workerChan <- struct{}{}:
-			go func() {
-				defer func() {
-					<-c.workerChan
-				}()
+			func() {
 
 				if c.usePriority && !c.priority.IsLatest(c.ctx) {
 					time.Sleep(5 * time.Second)
@@ -572,7 +572,7 @@ func (c *Cluster) executeTasks() {
 					return
 				}
 
-				c.processTask(taskID[1])
+				go c.processTask(taskID[1])
 			}()
 		}
 	}
@@ -582,6 +582,10 @@ func (c *Cluster) executeTasks() {
 // 执行任务
 func (l *Cluster) processTask(taskId string) {
 
+	defer func() {
+		<-l.workerChan
+	}()
+
 	begin := time.Now()
 
 	ctx, cancel := context.WithTimeout(l.ctx, l.timeout)
@@ -618,7 +622,8 @@ func (l *Cluster) processTask(taskId string) {
 	}
 
 	// 这里加一个全局锁
-	lock, err := lockx.NewGlobalLock(ctx, l.redis, taskId)
+	lockKey := l.lockKey + "_task_" + taskId
+	lock, err := lockx.NewGlobalLock(ctx, l.redis, lockKey)
 	if err != nil {
 		l.logger.Errorf(ctx, "doTask timer:获取锁失败:%s", taskId)
 		return
diff --git a/keys.go b/keys.go
new file mode 100644
index 0000000..febf85e
--- /dev/null
+++ b/keys.go
@@ -0,0 +1,11 @@
+package timerx
+
+import "strings"
+
+// sanitizeKeyPrefix 移除 keyPrefix 中的 {} 字符，避免破坏 Redis Cluster hash tag
+// Redis Cluster 使用 first {..} 配对作为 hash tag，keyPrefix 中包含的 {} 会干扰此机制
+func sanitizeKeyPrefix(keyPrefix string) string {
+	keyPrefix = strings.ReplaceAll(keyPrefix, "{", "_")
+	keyPrefix = strings.ReplaceAll(keyPrefix, "}", "_")
+	return keyPrefix
+}
diff --git a/once.go b/once.go
index 4c1e1c7..fd7cd7a 100644
--- a/once.go
+++ b/once.go
@@ -106,6 +106,9 @@ func InitOnce(ctx context.Context, re redis.UniversalClient, keyPrefix string, c
 		return nil, errors.New("redis client is nil")
 	}
 
+	// 清理 keyPrefix 中的 {} 字符，避免破坏 Redis Cluster hash tag
+	keyPrefix = sanitizeKeyPrefix(keyPrefix)
+
 	ctx, cancel := context.WithCancel(ctx)
 
 	u, err := uuid.NewV7()
@@ -307,10 +310,7 @@ func (l *Once) executeTasks() {
 		case <-l.ctx.Done():
 			return
 		case l.workerChan <- struct{}{}:
-			go func() {
-				defer func() {
-					<-l.workerChan
-				}()
+			func() {
 
 				if l.usePriority && !l.priority.IsLatest(l.ctx) {
 					time.Sleep(time.Second * 5)
@@ -331,7 +331,7 @@ func (l *Once) executeTasks() {
 					return
 				}
 				// 处理任务
-				l.processTask(keys[1])
+				go l.processTask(keys[1])
 			}()
 		}
 	}
@@ -566,6 +566,10 @@ func (l *Once) batchGetTasks() {
 // 执行任务
 func (l *Once) processTask(key string) {
 
+	defer func() {
+		<-l.workerChan
+	}()
+
 	begin := time.Now()
 
 	ctx, cancel := context.WithTimeout(l.ctx, l.timeout)