Flutter实现WebSocket通道报错：Bad state: stream已被监听

核心原因 #

WebSocket的stream是单订阅流（Single-Subscription Stream），这类流仅允许被监听一次，当应用从后台恢复后你尝试再次订阅同一个已被监听过的流时，就会触发该错误。

解决方案 #

1. 转换为多订阅流 #

通过asBroadcastStream()将WebSocket的单订阅流转为多订阅流，支持多次监听：

// 建立WebSocket连接后，把流转为广播流
final webSocket = await WebSocket.connect('ws://your-server-url');
final broadcastStream = webSocket.stream.asBroadcastStream();

// 后续可多次监听该广播流
broadcastStream.listen((message) {
  // 处理消息逻辑
});

2. 严格管理订阅生命周期 #

在Bloc中维护WebSocket连接和订阅的引用，应用恢复时先检查状态，避免重复订阅：

class YourBloc extends Bloc<YourEvent, YourState> {
  WebSocket? _webSocket;
  StreamSubscription? _wsSubscription;

  YourBloc() : super(YourInitialState()) {
    on<ConnectWsEvent>(_onConnectWs);
    on<ResumeWsEvent>(_onResumeWs);
  }

  Future<void> _onConnectWs(ConnectWsEvent event, Emitter<YourState> emit) async {
    _disposeWs(); // 先清理旧连接
    _webSocket = await WebSocket.connect('ws://your-server-url');
    _startListening(emit);
  }

  Future<void> _onResumeWs(ResumeWsEvent event, Emitter<YourState> emit) async {
    if (_webSocket == null || _webSocket!.closeCode != null) {
      // 连接已关闭，重新创建并监听
      await _onConnectWs(event, emit);
    } else if (_wsSubscription == null || _wsSubscription!.isPaused) {
      // 订阅已暂停，恢复监听
      _wsSubscription?.resume();
    }
    // 订阅已活跃则不做操作
  }

  void _startListening(Emitter<YourState> emit) {
    _wsSubscription = _webSocket?.stream.listen(
      (message) => emit(YourDataReceivedState(message)),
      onError: (err) => emit(YourErrorState(err)),
      onDone: _disposeWs,
    );
  }

  void _disposeWs() {
    _wsSubscription?.cancel();
    _webSocket?.close();
    _wsSubscription = null;
    _webSocket = null;
  }

  @override
  Future<void> close() {
    _disposeWs();
    return super.close();
  }
}

3. 用BehaviorSubject封装流（结合RxDart） #

如果项目使用RxDart，可通过BehaviorSubject缓存最新数据并支持多订阅：

import 'package:rxdart/rxdart.dart';

class WsService {
  WebSocket? _webSocket;
  final _messageSubject = BehaviorSubject<String>();

  Stream<String> get messageStream => _messageSubject.stream;

  Future<void> connect() async {
    await disconnect();
    _webSocket = await WebSocket.connect('ws://your-server-url');
    _webSocket?.stream.listen(
      (msg) => _messageSubject.add(msg.toString()),
      onError: (err) => _messageSubject.addError(err),
      onDone: disconnect,
    );
  }

  Future<void> disconnect() async {
    _webSocket?.close();
    _webSocket = null;
    if (!_messageSubject.isClosed) _messageSubject.close();
  }
}

在Bloc中只需监听messageStream，应用恢复时重新调用connect()即可自动绑定新流。

验证要点 #

应用从后台恢复时：

• 禁止重复订阅同一单订阅流
• 优先检查现有连接状态，仅在必要时重建连接或恢复订阅
• 优先使用多订阅流或Subject处理多次监听场景

内容的提问来源于stack exchange，提问作者Vinay Saurabh