优化部分并发问题

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() {
+			func() {
-				defer func() {
-					<-c.workerChan
-				}()
 
 				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

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
+}

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() {
+			func() {
-				defer func() {
-					<-l.workerChan
-				}()
 
 				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)