为何Dart程序报“Bad state: Stream has already been listened to”错误？

我已经在StackOverflow及其他平台广泛搜索类似问题，但未找到针对我这个问题的解决方案。

我正在尝试使用Dart Streams进行事件处理，编写了两个相似但不完全相同的程序。第一个程序运行正常，第二个程序却触发了Bad state: Stream has already been listened to错误，我无法找出原因。

若您能提供分析思路，我将万分感激。

第一个程序（运行正常） #

import 'dart:async';

// Get a Stream as broadcast
StreamController<String> controller = StreamController<String>();
Stream<String> stream = controller.stream.asBroadcastStream();

void main(){
  // define listener 1
  StreamSubscription<String> subscriber = stream.listen((String data){
            print('Receiver1: $data');
      }, 
      onError: (error){
            print('Error occured: ${error}');
               },
      onDone: (){
            print('Stream closed');
  });

  // define listener 2
  StreamSubscription<String> subscriber2 = stream.listen((String data){
            print('Receiver2: $data');
      }, 
      onError: (error){
            print('Error occured: ${error}');
               },
      onDone: (){
            print('Stream closed');
  });

  // in every second emit an event. On 3rd second emit an error
  for(int i=0; i<5; i++){
    Timer(Duration(seconds: i), (){ 
      if (i == 3){
        controller.sink.addError(ArgumentError.new('important error'));
        } else {
          controller.sink.add('Seconds passed: $i');
        }
    });
  }
  print('finished');
}

第二个程序（触发错误） #

import 'dart:async';

// Intention-specific StreamController
class MyStream  extends ControllerBase {}

// Just a StreamControllerBase, which centeralize the general idea
abstract class ControllerBase {
  final StreamController _streamController = StreamController();
  Stream<dynamic> get asBroadcastStream =>
      _streamController.stream.asBroadcastStream();
  StreamController get ctr => _streamController;

  void dispose() {
    _streamController.close();
  }
}

// General registry of StreamControllers
class ControllerProvider {
  static final ControllerProvider _controllerProvider =
      ControllerProvider._internal();

  // private constructor
  ControllerProvider._internal();

  // factory method
  factory ControllerProvider() {
    return _controllerProvider;
  }

  final Map<Type, dynamic> _registry = {};

  void register<T>(T controller) {
    _registry[T] = controller;
  }

  T get<T>() {
    return _registry[T] as T;
  }
}


// Just a class, which acts on MyStream
class B{
  B(){
    Stream stream = ControllerProvider().get<MyStream>().asBroadcastStream;
    stream.listen((data){
      print('Aha! from class B: ${data as String}');
    });
  }
}

// Another simple class, which also acts on MyStream. This class has an
// instance of class B. 
class A{
  B _b = B();
  A(){
    Stream stream = ControllerProvider().get<MyStream>().asBroadcastStream;
    stream.listen((data){
      print('Gotcha from class A: ${data as String}');
    });
  }
}


void main(){
  // Get an instance of MyStream
  MyStream myStream = MyStream();
  // register myStream instance in ControllerProvider
  ControllerProvider cp = ControllerProvider();
  cp.register<MyStream>(myStream);

  // Instantiate class A (both class A and class B instances should listen now)
  A a = A();

  // emit an event
  myStream.ctr.sink.add('helloo ');
  print('finished');
}

问题原因分析 #

问题核心在ControllerBase类的asBroadcastStream getter方法：

Stream<dynamic> get asBroadcastStream =>
    _streamController.stream.asBroadcastStream();

每次调用这个getter，都会基于原单订阅流（_streamController.stream默认是单订阅流）创建新的广播流实例。而原单订阅流只能被监听一次，当类B和类A分别调用该getter并监听时，相当于两次尝试绑定原单订阅流，直接触发报错。

反观第一个正常程序，它提前创建唯一的广播流实例，后续所有监听都复用这个实例：

Stream<String> stream = controller.stream.asBroadcastStream();

广播流本身支持多订阅，因此两次监听不会冲突。

解决方案 #

修改ControllerBase类，确保只创建一次广播流实例，后续复用该实例：

abstract class ControllerBase {
  final StreamController _streamController = StreamController();
  late final Stream<dynamic> _broadcastStream;

  ControllerBase() {
    _broadcastStream = _streamController.stream.asBroadcastStream();
  }

  Stream<dynamic> get asBroadcastStream => _broadcastStream;
  StreamController get ctr => _streamController;

  void dispose() {
    _streamController.close();
  }
}

这样无论多少次调用asBroadcastStream，返回的都是同一个广播流实例，支持多订阅，即可解决报错问题。

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