修复一个任务调度BUG

cluster.go

@@ -541,9 +541,14 @@ func (c *Cluster) executeTasks() {
 		case <-c.ctx.Done():
 			return
 		case c.workerChan <- struct{}{}:
+			func() {
+				defer func() {
+					<-c.workerChan
+				}()
+
 				if c.usePriority && !c.priority.IsLatest(c.ctx) {
 					time.Sleep(5 * time.Second)
-				continue
+					return
 				}
 
 				taskID, err := c.redis.BLPop(c.ctx, 10*time.Second, c.listKey).Result()
@@ -553,16 +558,17 @@ func (c *Cluster) executeTasks() {
 						// Redis 异常，休眠一会儿
 						time.Sleep(5 * time.Second)
 					}
-				continue
+					return
 				}
 
 				if len(taskID) < 2 {
 					c.logger.Errorf(c.ctx, "Invalid BLPop result: %v", taskID)
 					// 数据异常，继续下一个
-				continue
+					return
 				}
 
 				go c.processTask(taskID[1])
+			}()
 		}
 	}
 
@@ -575,9 +581,6 @@ type ReJobData struct {
 
 // 执行任务
 func (l *Cluster) processTask(taskId string) {
-	defer func() {
-		<-l.workerChan
-	}()
 
 	begin := time.Now()
 

once.go

@@ -289,19 +289,21 @@ 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
+				}()
+
+				if l.usePriority && !l.priority.IsLatest(l.ctx) {
+					time.Sleep(time.Second * 5)
+					return
+				}
 
 				keys, err := l.redis.BLPop(l.ctx, time.Second*10, l.listKey).Result()
 				if err != nil {
@@ -310,17 +312,17 @@ func (l *Once) executeTasks() {
 						// Redis 异常，休眠一会儿再重试
 						time.Sleep(time.Second * 5)
 					}
-				continue
+					return
 				}
 
 				if len(keys) < 2 {
 					l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
 					// 数据异常，继续下一个
-				continue
+					return
 				}
 				// 处理任务
 				go l.processTask(keys[1])
-
+			}()
 		}
 	}
 
@@ -518,9 +520,6 @@ func (l *Once) batchGetTasks() {
 
 // 执行任务
 func (l *Once) processTask(key string) {
-	defer func() {
-		<-l.workerChan
-	}()
 
 	begin := time.Now()