优化

internal/worker/task_scheduler.go

@@ -55,7 +55,7 @@ func (s *TaskScheduler) Start(ctx context.Context) error {
 				}
 			}
 		}
-		// GVA reports the number of rows read from the database, not only
+		// Report the number of rows read from the database, not only
 		// the enabled rows that were scheduled successfully.
 		s.logger.InfoContext(ctx, "定时任务加载完成", "task_count", len(items))
 	} else {
@@ -222,6 +222,13 @@ func (s *TaskScheduler) Subscribe(userID uint) chan []byte {
 	if s.subscribers[userID] == nil {
 		s.subscribers[userID] = map[chan []byte]struct{}{}
 	}
+	if len(s.subscribers[userID]) >= 10 {
+		for old := range s.subscribers[userID] {
+			delete(s.subscribers[userID], old)
+			close(old)
+			break
+		}
+	}
 	s.subscribers[userID][ch] = struct{}{}
 	s.subMu.Unlock()
 	return ch

internal/worker/task_scheduler_test.go

@@ -23,3 +23,28 @@ func TestCloseSubscribersClosesEveryAlertStream(t *testing.T) {
 	// closed and removed every channel.
 	scheduler.Unsubscribe(1, first)
 }
+
+func TestSubscribeCapsConnectionsPerUser(t *testing.T) {
+	scheduler := &TaskScheduler{subscribers: map[uint]map[chan []byte]struct{}{}}
+	channels := make([]chan []byte, 0, 11)
+	for i := 0; i < 11; i++ {
+		channels = append(channels, scheduler.Subscribe(7))
+	}
+	if got := len(scheduler.subscribers[7]); got != 10 {
+		t.Fatalf("subscriber count = %d, want 10", got)
+	}
+	closed := 0
+	for _, ch := range channels {
+		select {
+		case _, open := <-ch:
+			if !open {
+				closed++
+			}
+		default:
+		}
+	}
+	if closed != 1 {
+		t.Fatalf("evicted subscriber count = %d, want 1", closed)
+	}
+	scheduler.closeSubscribers()
+}