一、问题的本质

写 Flutter 开发，Stream 是躲不掉的。不管是 StreamBuilder、BLoC 模式，还是 Riverpod，亦或是 TextField 的 onChanged 回调，背后全都在跑 Stream。

但很多开发者平时只是调用，一旦面试官抛出一个深层问题，现场往往就安静了。比如：

“请解释一下单订阅 Stream 和广播 Stream 的区别？如果我要从头实现一个 StreamController，或者用 async* 写一个生成器函数，流程是怎样的？StreamTransformer 又是怎么工作的？”

这个问题其实包含了五个层级。初级开发者只能答出第一层，而高级工程师需要能够把这些概念串联起来，并结合实际的架构决策（比如为什么在 BLoC 中要这么设计）来回答。今天我不讲那些虚的，直接拆解整个 Stream 的心智模型。

二、Stream 的核心机制：单订阅与广播

这是 Flutter 面试中最常问的题目，如果不搞清楚这个，你大概率会遇到 Bad state: Stream has already been listened to 这个报错，然后对着屏幕抓耳挠腮好几个小时。

1、单订阅 vs 广播：到底有什么区别？

我们直接看对比表格，这样更直观：

特性	单订阅 Stream (Single-subscription)	广播 Stream (Broadcast)
订阅者数量	只能有一个订阅者	可以有多个订阅者同时监听
未监听时的行为	会缓存数据，等待第一个订阅者出现	直接丢弃数据，没人在听就白说了
重复监听后果	抛出 StateError	没有任何问题
典型使用场景	文件读取、单次 HTTP 请求	UI 事件、BLoC 状态流、WebSocket

下面通过一段代码来看看这个报错是怎么发生的：

import 'dart:async';

void main() async {
  // 默认是单订阅 Stream
  final controller = StreamController<int>();

  controller.sink.add(1);
  controller.sink.add(2);
  controller.close();

  // 第一个监听者：一切正常
  controller.stream.listen(
    (v) => print('Listener A: $v'),
    onDone: () => print('A done'),
  );

  // 第二个监听者：直接报错！
  // controller.stream.listen((v) => print('Listener B: $v'));
  // ✗ Error: Bad state: Stream has already been listened to
}

2、如何解决“重复监听”的报错？

如果你手里拿的是一个单订阅 Stream，但业务逻辑需要多个组件（比如两个 StreamBuilder）同时监听它，最简单的办法就是用 .asBroadcastStream() 将其转换成广播流。

Stream<int> singleStream() async* {
  yield 1;
  yield 2;
}

void main() async {
  // 转换成广播流，这样多个 Widget 就能共用了
  final broadcast = singleStream().asBroadcastStream();

  broadcast.listen((v) => print('Widget 1: $v'));
  broadcast.listen((v) => print('Widget 2: $v'));
}

三、进阶玩法：从生成器到自定义转换器

当你不再满足于简单的监听，而是想要主动 产生 数据或者对数据进行 加工 时，就需要用到生成器和转换器了。

1、async* 与 yield 的巧妙配合

async* 函数就是一个“生成器”（Generator），它可以懒加载地产生一个 Stream。你不需要手动去操作 StreamController，只需要用 yield 把值“扔”出来即可。

这里有个细节：yield 是产生一个值，而 yield* 则是“委托”另一个流。

import 'dart:async';

// 使用 async* 生成流
Stream<int> countdown(int from) async* {
  for (int i = from; i >= 0; i--) {
    await Future.delayed(const Duration(milliseconds: 100));
    yield i; // 产出一个值，然后暂停，等监听者准备好再继续
  }
}

// 使用 yield* 委托另一个流
Stream<int> combined() async* {
  yield* countdown(3); // 把 countdown 的结果直接“拼”过来
  yield 100;
}

void main() async {
  await for (final n in combined()) {
    print(n);
  }
}

2、StreamTransformer：构建你的异步流水线

如果你觉得 map 或 where 这种简单的操作符不够用，想要实现更复杂的逻辑（比如防抖 debounce、日志记录或格式转换），那就得祭出 StreamTransformer 了。它就像是流水线上的加工环节，把 Stream<S> 转换成 Stream<T>。

// 一个简单的日志转换器，方便在开发时观察流的变化
StreamTransformer<T, T> logged<T>(String tag) {
  return StreamTransformer.fromHandlers(
    handleData: (data, sink) {
      print('[$tag] data: $data');
      sink.add(data);
    },
    handleError: (e, st, sink) {
      print('[$tag] error: $e');
      sink.addError(e, st);
    },
    handleDone: (sink) {
      print('[$tag] done');
      sink.close();
    },
  );
}

四、架构与实战：从 BLoC 到 StreamBuilder

最后，我们来看如何把这些零散的零件组装成真正的架构。

1、StreamController：手动控制流的生命周期

在构建像 BLoC 这种需要高度控制权的架构时，StreamController 是核心。它提供了 sink 用于输入（add），提供了 stream 用于输出。

这里必须强调：一定要手动关闭！ 无论是在 StatefulWidget 的 dispose() 里，还是在 BLoC 的 close() 里，不释放 StreamController 是 Flutter 内存泄漏的最主要诱因。

2、StreamBuilder：连接流与 UI 的桥梁

StreamBuilder 是 Flutter 中连接异步数据与 Widget 树的纽带。它通过 AsyncSnapshot 来展示当前流的状态。你得搞清楚 ConnectionState 的几种状态：

• none: 没有提供流。

• waiting: 正在监听，但还没拿到数据。

• active: 正在稳定输出数据。

• done: 流已经关闭。

3、手写一个最简 BLoC

BLoC 的本质就是：接收一个事件流（Events），经过逻辑处理，输出一个状态流（States）。

import 'dart:async';

// 事件定义
abstract class CounterEvent {}
class Increment extends CounterEvent {}

// BLoC 实现
class CounterBloc {
  int _count = 0;

  // 输入：UI 触发事件
  final _eventController = StreamController<CounterEvent>();
  Sink<CounterEvent> get events => _eventController.sink;

  // 输出：UI 监听状态
  final _stateController = StreamController<int>.broadcast();
  Stream<int> get state => _stateController.stream;

  CounterBloc() {
    _eventController.stream.listen(_handleEvent);
  }

  void _handleEvent(CounterEvent event) {
    if (event is Increment) _count++;
    _stateController.add(_count);
  }

  void dispose() {
    _eventController.close();
    _stateController.close();
  }
}

五、总结：面试避坑指南

既然是深度解析，最后我把这次“折腾”总结成几个面试高频点，方便大家直接理解消化：

1. 单订阅 vs 广播：单订阅只能听一次，听完就废；广播可以听多次，但没人在听时数据就丢了。

2. async* 与 yield：async* 是生成器，yield 是产出，yield* 是嵌套。

3. 内存泄漏：记得在 dispose 里关闭 StreamController，并取消（cancel）你的 StreamSubscription。

4. StreamBuilder 陷阱：不要在 build 方法里直接实例化 Stream，否则每次 rebuild 都会创建一个新流，导致性能问题甚至逻辑错误。你应该在 initState 里初始化它。

掌握了这些，你才算真正理解了 Dart 的响应式灵魂。