稳定发布1.0.0

internal/routes/wait.go

@@ -31,6 +31,9 @@ func runAction(c *leaf.Context, item schema.WaitItemModel, rootRules []cronrange
 		return
 	}
 
+	a := make(chan bool)
+	defer close(a)
+
 	// 等待组
 	var wait sync.WaitGroup
 	defer wait.Wait()
@@ -38,46 +41,57 @@ func runAction(c *leaf.Context, item schema.WaitItemModel, rootRules []cronrange
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 
-	ctx, cancel := context.WithCancel(context.TODO())
-
-	var run bool
-
 	wait.Add(1)
 	go func() {
 		defer wait.Done()
 		for {
 			select {
 			case <-c.Done():
-				cancel()
 				return
 			case <-ticker.C:
 				if cronrange.Match(rules, time.Now()) && cronrange.Match(rootRules, time.Now()) {
-					run = true
+					a <- true
 				} else {
-					run = false
-					cancel()
+					a <- false
 				}
 			}
 		}
 	}()
 
+	var cancel context.CancelFunc
+	var m sync.Mutex
 	for {
 		select {
 		case <-c.Done():
+			if cancel != nil {
+				cancel()
+			}
 			return
-		default:
-			if run {
-				err := play(ctx, item)
-				if err != nil {
-					zap.S().Errorln("执行动作异常: ", err)
-					<-time.After(time.Minute)
-				}
-			} else {
-				select {
-				case <-c.Done():
-					return
-				case <-time.After(time.Second):
+		case r := <-a:
+			if r {
+				if ok := m.TryLock(); ok {
+					ctx, cc := context.WithCancel(context.TODO())
+					cancel = cc
+					wait.Add(1)
+					go func() {
+						defer wait.Done()
+						defer m.Unlock()
+						defer func() { cancel = nil }()
+
+						err := play(ctx, item)
+						if err != nil {
+							zap.S().Errorln("执行动作异常: ", err)
+							select {
+							case <-ctx.Done():
+								return
+							case <-time.After(time.Minute):
+							}
+						}
+					}()
 				}
+			} else if cancel != nil {
+				cancel()
+				cancel = nil
 			}
 		}
 	}