优化限制并发的生效

cluster.go

@@ -550,7 +550,7 @@ func (c *Cluster) executeTasks() {
 			return
 		case <-c.ctx.Done():
 			return
-		case c.workerChan <- struct{}{}:
+		default:
 			func() {
 
 				if c.usePriority && !c.priority.IsLatest(c.ctx) {
@@ -572,6 +572,9 @@ func (c *Cluster) executeTasks() {
 					return
 				}
 
+				// 限制并发数量
+				c.workerChan <- struct{}{}
+
 				go c.processTask(taskID[1])
 			}()
 		}

once.go

@@ -309,7 +309,7 @@ func (l *Once) executeTasks() {
 			return
 		case <-l.ctx.Done():
 			return
-		case l.workerChan <- struct{}{}:
+		default:
 			func() {
 
 				if l.usePriority && !l.priority.IsLatest(l.ctx) {
@@ -330,7 +330,10 @@ func (l *Once) executeTasks() {
 					l.logger.Errorf(l.ctx, "Invalid task data: %v", keys)
 					return
 				}
-				// 处理任务
+
+				// 限制并发数量
+				l.workerChan <- struct{}{}
+
 				go l.processTask(keys[1])
 			}()
 		}