颤振：坏状态：流已经被收听

Question

    class MyPage extends StatelessWidget {
      @override
      Widget build(BuildContext context) {
        return DefaultTabController(
          length: 2,
          child: new Scaffold(
            appBar: TabBar(
              tabs: [
                Tab(child: Text("MY INFORMATION",style: TextStyle(color: Colors.black54),)),
                Tab(child: Text("WEB CALENDER",style: TextStyle(color: Colors.black54),)),
              ],
            ),
            body:PersonalInformationBlocProvider(
              movieBloc: PersonalInformationBloc(),
              child: TabBarView(
                children: [
                  MyInformation(),
                  new SmallCalendarExample(),
                ],
              ),
            ),
          ),
        );
      }
    }
    
    class MyInformation extends StatelessWidget{
      // TODO: implement build
      var deviceSize;
    
      //Column1
      Widget profileColumn(PersonalInformation snapshot) => Container(
        height: deviceSize.height * 0.24,
        child: Column(
          mainAxisAlignment: MainAxisAlignment.center,
          children: <Widget>[
            Row(
              mainAxisAlignment: MainAxisAlignment.spaceEvenly,
              children: <Widget>[
                Container(
                  decoration: BoxDecoration(
                    borderRadius:
                    new BorderRadius.all(new Radius.circular(50.0)),
                    border: new Border.all(
                      color: Colors.black,
                      width: 4.0,
                    ),
                  ),
                  child: CircleAvatar(
                    backgroundImage: NetworkImage(
                        "http://www.binaythapa.com.np/img/me.jpg"),
                    foregroundColor: Colors.white,
                    backgroundColor: Colors.white,
                    radius: 40.0,
                  ),
                ),
                ProfileTile(
                  title: snapshot.firstName,
                  subtitle: "Developer",
                ),
                SizedBox(
                  height: 10.0,
                ),
              ],
            )
          ],
        ),
      );
      Widget bodyData(PersonalInformation snapshot) {
        return SingleChildScrollView(
            child: Column(
              children: <Widget>[
                profileColumn(snapshot)
              ],
            ),
        );
      }
    
    
      @override
      Widget build(BuildContext context) {
        final personalInformationBloc = PersonalInformationBlocProvider.of(context);
    
        deviceSize = MediaQuery.of(context).size;
        return StreamBuilder(
            stream: personalInformationBloc.results,
            builder: (context,snapshot){
              if (!snapshot.hasData)
                return Center(
                  child: CircularProgressIndicator(),
                );
              return bodyData(snapshot.data);
            }
        );
      }
    }
   

我正在使用 Bloc 模式从 Rest API 检索数据（仅从 JSON 调用整个对象并仅解析用户名）。该页面由两个选项卡 MyInformation 和 SmallCalendar 组成。当应用程序运行时，数据被正确获取并且一切都很好。当我转到标签二并返回标签一时，标签一中的整个屏幕变为红色，显示错误： Bad state: Stream has already been listened to.

Answer 1

我在导航离开然后返回到监听流的视图时遇到此错误，因为我将同一视图的新实例推送到导航器堆栈中，这实际上最终创建了一个新的监听器，即使它是代码中的相同位置。

更具体地说，我有一个ListItemsView 小部件，它使用StreamBuilder 来显示流中的所有项目。用户点击“添加项目”按钮，该按钮将AddItemView 推送到导航器堆栈中，提交表单后，用户被带回ListItemsView，其中“错误状态：流已被收听。 "发生错误。

对我来说，解决方法是将Navigator.pushNamed(context, ListItemsView.routeName) 替换为Navigator.pop(context)。这有效地防止了新 ListItemsView 的实例化（作为同一流的第二个订阅者），并且只是将用户带回之前的 ListItemsView 实例。

Answer 2

确保您处置了控制器！

@override
  void dispose() {
    scrollController.dispose();
    super.dispose();
  }

Answer 3

在我的情况下，我收到此错误是因为同一行代码 myStream.listen() 在同一流的同一小部件​​中被调用了两次。显然这是不允许的！

更新： 如果您打算多次订阅同一个流，则应使用行为主题：

// 1- Create a behavior subject
final _myController = BehaviorSubject<String>();

// 2- To emit/broadcast new events, we will use Sink of the behavior subject.
Sink<String> get mySteamInputSink => _myController.sink;

// 3- To listen/subscribe to those emitted events, we will use Stream (observable) of the behavior subject. 
Stream<String> get myStream => _myController.stream;

// 4- Firstly, Listen/subscribe to stream events.
myStream.listen((latestEvent) {
   // use latestEvent data here.
});

// 5- Emit new events by adding them to the BehaviorSubject's Sink. 
myStreamInputSink.add('new event');

就是这样！

但是，还有最后一个重要步骤。

6- 在销毁小部件之前，我们必须取消订阅所有流侦听器。

为什么？ （你可能会问）

因为如果一个小部件订阅了一个流，并且当这个小部件被销毁时，被销毁的小部件流订阅将保留在应用程序内存中，从而导致内存泄漏和不可预知的行为。：

_flush() {
  _myController.close();
  _myController = StreamController<String>();
}

############################### ###############################

旧答案：

为我解决的问题是将我的流控制器创建为广播流控制器：

var myStreamController = StreamController<bool>.broadcast();

与

使用流作为广播流：

myStreamController.stream.asBroadcastStream().listen(onData);

Answer 4

我认为并非所有答案都考虑到您不想或根本无法使用广播流的情况。

通常情况下，您必须依赖接收过去的事件，因为侦听器的创建时间可能晚于它所侦听的流，因此接收此类信息很重要。

在 Flutter 中，经常发生的情况是侦听流的小部件（“侦听器”）被销毁并重新构建。如果您尝试将侦听器附加到与以前相同的流，您将收到此错误。

要克服这个问题，您必须手动管理流。我创建了this gist 来演示如何做到这一点。您还可以在 this dartpad 上运行此代码，以查看它的行为方式并使用它。我使用简单的String id 来引用特定的StreamController 实例，但也可能有更好的解决方案（也许是symbols）。

要点的代码是：

/* NOTE: This approach demonstrates how to recreate streams when
         your listeners are being recreated.
         It is useful when you cannot or do not want to use broadcast
         streams. Downside to broadcast streams is that it is not
         guaranteed that your listener will receive values emitted
         by the stream before it was registered.
*/

import 'dart:async';
import 'dart:math';

// [StreamService] manages state of your streams. Each listener
// must have id which is used in [_streamControllers] map to
// look up relevant stream controller.
class StreamService {
  final Map<String, StreamController<int>?> _streamControllers = {};

  Stream<int> getNamedStream(String id) {
    final controller = _getController(id);
    return controller.stream;
  }

  // Will get existing stream controller by [id] or create a new
  // one if it does not exist
  StreamController<int> _getController(String id) {
    final controller = _streamControllers[id] ?? _createController();

    _streamControllers[id] = controller;

    return controller;
  }

  void push(String id) {
    final controller = _getController(id);

    final rand = Random();
    final value = rand.nextInt(1000);

    controller.add(value);
  }

  // This method can be called by listener so
  // memory leaks are avoided. This is a cleanup
  // method that will make sure the stream controller
  // is removed safely
  void disposeController(String id) {
    final controller = _streamControllers[id];

    if (controller == null) {
      throw Exception('Controller $id is not registered.');
    }

    controller.close();
    _streamControllers.remove(id);
    print('Removed controller $id');
  }

  // This method should be called when you want to remove
  // all controllers. It should be called before the instance
  // of this class is garbage collected / removed from memory.
  void dispose() {
    _streamControllers.forEach((id, controller) {
      controller?.close();
      print('Removed controller $id during dispose phase');
    });
    _streamControllers.clear();
  }

  StreamController<int> _createController() {
    return StreamController<int>();
  }
}

class ManagedListener {
  ManagedListener({
    required this.id,
    required StreamService streamService,
  }) {
    _streamService = streamService;
  }

  final String id;
  late StreamService _streamService;
  StreamSubscription<int>? _subscription;

  void register() {
    _subscription = _streamService.getNamedStream(id).listen(_handleStreamChange);
  }

  void dispose() {
    _subscription?.cancel();
    _streamService.disposeController(id);
  }

  void _handleStreamChange(int n) {
    print('[$id]: streamed $n');
  }
}

void main(List<String> arguments) async {
  final streamService = StreamService();

  final listener1Id = 'id_1';
  final listener2Id = 'id_2';

  final listener1 = ManagedListener(id: listener1Id, streamService: streamService);
  listener1.register();
  
  streamService.push(listener1Id);
  streamService.push(listener1Id);
  streamService.push(listener1Id);

  await Future.delayed(const Duration(seconds: 1));

  final listener2 = ManagedListener(id: listener2Id, streamService: streamService);
  listener2.register();

  streamService.push(listener2Id);
  streamService.push(listener2Id);

  await Future.delayed(const Duration(seconds: 1));
  
  listener1.dispose();
  listener2.dispose();

  streamService.dispose();
}

Answer 5

对于其他情况。请注意您是否以某种方式在无状态类中使用流。这是您收到上述错误的原因之一。 将无状态类转换为有状态并在streamController上调用init和dispose方法：

 @override
 void initState() {
   super.initState();
   YourStreamController.init();
 }

 @override
 void dispose() {
   YourStreamController.dispose();
   super.dispose();
 }

Answer 6

这可以帮助任何其他人。就我而言，我在每个选项卡中使用了两个 StreamBuilder。因此，当我滑动到选项卡上方并返回时。另一个流已经被监听，所以我得到了错误。

我所做的是从选项卡中删除 StreamBuilder 并将其放在顶部。每次有变化时我都会设置状态。我返回一个空的 Text('') 以避免显示任何内容。我希望这种方法

Answer 7

这是提供者的问题，我通过更改提供者初始化解决了它

例如

locator.registerSingleton<LoginProvider>(LoginProvider());

到

 locator.registerFactory(() => TaskProvider());

定位器在哪里

GetIt locator = GetIt.instance;

Answer 8

您可以使用broadcast，它允许多次监听流，但它也阻止监听过去的事件：

没有监听器时，广播流不会缓冲事件。

更好的选择是使用来自rxdart 包类的BehaviorSubject 作为StreamController。 BehaviorSubject 是：

一个特殊的 StreamController 捕获已添加到控制器的最新项目，并将其作为第一个项目发送给任何新的侦听器。

用法很简单：

StreamController<...> _controller = BehaviorSubject();

Answer 9

在我的例子中，我在 Flutter Web 上使用了 Package Connectivity。 评论所有连接调用解决了这个问题。

我现在只在 Android/iOS 上使用 Connectivity。

因此，如果您正在为 Web 开发，请检查您的包，因为您正在使用一些在 Web 上存在问题的包。

希望我能帮助有关此信息的人。

Answer 10

当我将 Observable.combineLatest2 的结果用于 StreamBuilder 到 Drawer 时，我遇到了同样的问题：

flutter：状态不佳：Stream 已被监听。

就我而言，最好的解决方案是将这种组​​合的结果添加到新的 BehaviorSubject 并监听新的。

别忘了听老歌！！！

class VisitsBloc extends Object {
    Map<Visit, Location> visitAndLocation;

    VisitsBloc() {
        visitAndLocations.listen((data) {
            visitAndLocation = data;
        });
    }

    final _newOne = new BehaviorSubject<Map<Visit, Location>>();

    Stream<Map<Visit, Location>> get visitAndLocations => Observable.combineLatest2(_visits.stream, _locations.stream, (List<vis.Visit> visits, Map<int, Location> locations) {
        Map<vis.Visit, Location> result = {};

        visits.forEach((visit) {
            if (locations.containsKey(visit.skuLocationId)) {
                result[visit] = locations[visit.skuLocationId];
            }
        });

        if (result.isNotEmpty) {
            _newOne.add(result);
        }
    });
}

我没有使用.broadcast，因为它减慢了我的 UI。

Answer 11

async 中的StreamSplitter.split() 可用于此用例

import 'package:async/async.dart';
...

main() {
  var process = Process.start(...);
  var stdout = StreamSplitter<List<int>>(process.stdout);
  readStdoutFoo(stdout.split());
  readStdoutBar(stdout.split());
}

readStdoutFoo(Stream<List<int>> stdout) {
  stdout.transform(utf8.decoder)...
}

readStdoutBar(Stream<List<int>> stdout) {
  stdout.transform(utf8.decoder)...
}

Answer 12

对我来说，将我的流定义为全局变量是可行的

流信息流（在有状态小部件的 ...State 内，我在小部件外部定义它并且它工作

（不确定是否是最佳解决方案，但请尝试一下）

Answer 13

对于那些在执行Future.asStream() 时遇到此问题的人，您需要Future.asStream().shareReplay(maxSize: 1) 将其设为广播/热流。

Answer 14

总结一下：

主要区别在于broadcast()创建Stream可用于多个来源，但它需要至少有一个来源才能开始发射项目。

在订阅者开始监听它之前，Stream 应该是惰性的（使用 [onListen] 回调开始产生事件）。

asBroadcastStream 将现有的 Stream 转换为可多听的，但它不需要被监听即可开始发射，因为它在后台调用 onListen()。

Answer 15

您应该使用以下内容。

StreamController<...> _controller = StreamController<...>.broadcast();

Answer 16

问题是由于没有在 bloc 中处理控制器。

  void dispose() {
    monthChangedController.close();
    dayPressedController.close();
    resultController.close();
  }

Answer 17

Stream 最常见的形式一次只能收听一次。如果你尝试添加多个监听器，它会抛出

Bad state: Stream 已经被监听

为防止出现此错误，请公开广播Stream。您可以使用myStream.asBroadcastStream将您的流转换为广播

这需要在暴露Stream 的类中完成。不是StreamBuilder 的参数。由于asBroadcastStream 在内部侦听原始流以生成广播流，这意味着您不能在同一流上调用此方法两次。