fix: 窗口移除 close(outputChan) 消除关闭期 send-on-closed panic

window/performance_test.go

@@ -77,9 +77,17 @@ func BenchmarkTumblingWindowThroughput(b *testing.B) {
 	go tw.Start()
 
 	// 在后台消费结果，避免阻塞
+	done := make(chan struct{})
 	go func() {
-		for range tw.OutputChan() {
-			// 消费结果
+		for {
+			select {
+			case _, ok := <-tw.OutputChan():
+				if !ok {
+					return
+				}
+			case <-done:
+				return
+			}
 		}
 	}()
 
@@ -104,6 +112,7 @@ func BenchmarkTumblingWindowThroughput(b *testing.B) {
 	b.Logf("发送成功: %d, 丢弃: %d", stats["sentCount"], stats["droppedCount"])
 
 	tw.Stop()
+	close(done)
 }
 
 // TestWindowBufferOverflow 测试缓冲区溢出处理

window/session_window.go

@@ -243,9 +243,6 @@ func (sw *SessionWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (sw *SessionWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
-
 		// Wait for initialization completion or context cancellation
 		select {
 		case <-sw.initChan:
@@ -283,8 +280,6 @@ func (sw *SessionWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (sw *SessionWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
 		if sw.watermark != nil {
 			defer sw.watermark.Stop()
 		}

window/sliding_window.go

@@ -272,9 +272,6 @@ func (sw *SlidingWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (sw *SlidingWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
-
 		// Wait for initialization complete or context cancellation
 		select {
 		case <-sw.initChan:
@@ -367,8 +364,6 @@ func (sw *SlidingWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (sw *SlidingWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(sw.outputChan)
 		if sw.watermark != nil {
 			defer sw.watermark.Stop()
 		}

window/tumbling_window.go

@@ -333,9 +333,6 @@ func (tw *TumblingWindow) Start() {
 // startProcessingTime starts the processing time trigger mechanism
 func (tw *TumblingWindow) startProcessingTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(tw.outputChan)
-
 		// Wait for initialization complete or context cancellation
 		select {
 		case <-tw.initChan:
@@ -381,8 +378,6 @@ func (tw *TumblingWindow) startProcessingTime() {
 // startEventTime starts the event time trigger mechanism based on watermark
 func (tw *TumblingWindow) startEventTime() {
 	go func() {
-		// Close output channel when function ends
-		defer close(tw.outputChan)
 		if tw.watermark != nil {
 			defer tw.watermark.Stop()
 		}