Flutter中StreamBuilder+WebSocket报错：Bad state: Stream已被监听

问题描述 #

通过WebSocket获取响应，基于该WebSocket创建Stream并赋值给StreamBuilder的stream属性来获取数据快照。但通过底部导航栏重新导航到使用该StreamBuilder的页面时，出现红屏错误：

Bad state: Stream has already been listened to

注：未使用任何StreamController，仅用StreamBuilder，通过底部导航栏切换页面。

关联代码：

import 'package:websocket_sample_app/modals/websocket_response.dart';
import 'package:flutter/material.dart';

import '../modals/data_provider.dart';

class CoinValue extends StatefulWidget {
  final WebsocketProvider provider;

  CoinValue({required this.provider});
  @override
  State<CoinValue> createState() => _CoinValueState();
}

class _CoinValueState extends State<CoinValue> {
  @override
  Widget build(BuildContext context) {
    return DataStream(stream: widget.provider.WebsocketStream);
  }
}

class DataStream extends StatefulWidget {
  final Stream<WebsocketResponse> stream;

  const DataStream({Key? key, required this.stream}) : super(key: key);

  @override
  State<DataStream> createState() => _DataStreamState();
}

class _DataStreamState extends State<DataStream> {
  @override
  Widget build(BuildContext context) {
    return StreamBuilder(
      stream: widget.stream,
      builder: (context, snapshot) {
        print("State of Connection:${(snapshot.connectionState).toString()}");
        if (snapshot.connectionState == ConnectionState.waiting) {
          return Center(
            child: CircularProgressIndicator(),
          );
        }
        print(
            "State of connection:${snapshot.connectionState == ConnectionState.active && snapshot.hasData}");
        if (snapshot.connectionState == ConnectionState.active &&
            snapshot.hasData) {
          WebsocketResponse? obj = snapshot.data as WebsocketResponse?;
          final response = obj!.toJson();
          return Text(
            '${response}',
            style: TextStyle(color: Colors.black),
          );
        }

        if (snapshot.connectionState == ConnectionState.done) {
          return const Center(
            child: Text(
              'No more data',
              style: TextStyle(
                color: Colors.red,
              ),
            ),
          );
        }

        return const Center(
          child: Text('No data'),
        );
      },
    );
  }
}

问题原因 #

普通的Stream是单订阅流（Single-Subscription Stream），只能被监听一次。当页面切换后重建，StreamBuilder会尝试再次监听同一个已被订阅过的流，就会抛出上述错误。

解决方法 #

将单订阅流转换为多订阅流（Broadcast Stream），同一个广播流可以被多次监听。有两种实现方式：

方式1：在WebsocketProvider中直接创建广播流 #

修改WebsocketProvider中的WebsocketStream属性，将原始流转换为广播流：

// 示例：在WebsocketProvider内部处理
class WebsocketProvider {
  Stream<WebsocketResponse> get WebsocketStream {
    // 这里是你原本创建WebSocket流的逻辑
    final originalStream = yourWebSocketConnectLogic();
    // 转换为广播流
    return originalStream.asBroadcastStream();
  }
}

方式2：在使用流时临时转换 #

如果无法修改WebsocketProvider，可以在传递给DataStream前转换：

class _CoinValueState extends State<CoinValue> {
  @override
  Widget build(BuildContext context) {
    // 将单订阅流转为广播流
    final broadcastStream = widget.provider.WebsocketStream.asBroadcastStream();
    return DataStream(stream: broadcastStream);
  }
}

额外注意事项 #

• 广播流不会缓存历史数据，新订阅者只能收到订阅后的新数据。如果需要缓存最新数据，可以结合BehaviorSubject（需引入rxdart库），或自行实现简单的缓存逻辑。
• 确保WebsocketProvider是全局单例（比如通过Provider、GetX等状态管理工具），避免页面切换时重复创建WebSocket连接。

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