fix: 关闭sink, 异步发包协程未退出

stream/sink.go

@@ -136,6 +136,8 @@ func (s *BaseSink) SetID(id SinkID) {
 }
 
 func (s *BaseSink) doAsyncWrite() {
+	defer s.existed.Store(true)
+
 	for {
 		select {
 		case <-s.cancelCtx.Done():
@@ -145,8 +147,6 @@ func (s *BaseSink) doAsyncWrite() {
 			break
 		}
 	}
-
-	s.existed.Store(true)
 }
 
 func (s *BaseSink) EnableAsyncWriteMode(queueSize int) {
@@ -263,6 +263,10 @@ func (s *BaseSink) Close() {
 			s.Conn.Close()
 			s.Conn = nil
 		}
+
+		if s.cancelCtx != nil {
+			s.cancelFunc()
+		}
 	}()
 
 	// 已经关闭或Sink未添加到任何队列, 不做处理