利用Stream实现Flutter的组件解耦

2024年7月4日 · 阅读需 5 分钟

Stream 是什么？

本质上，Stream 是一个异步数据队列，具有先进先出（First In First Out，FIFO）的特性。借助它，组件之间可以彻底解耦——数据生产者只管往流里推数据，消费者只关心如何订阅，两者互不感知，数据流动因此更加灵活、可控。

Stream 和 Future 的区别

简单说：Future 只返回一次结果，Stream 可以连续返回多个结果。

Stream 的分类

Stream 分为两种：

单订阅流（single-subscription）：只能有一个订阅者，也就是只能有一个消费者。
多订阅流（broadcast）：可以有多个订阅者，也就是可以有多个消费者。用 StreamController.broadcast() 创建，或对已有的流调用 .asBroadcastStream() 转换。

Stream 的使用

单订阅流的构造

import 'dart:async';

void main() {
  createSingleStream();
}

Future<void> createSingleStream() async {
  // 1) Stream.periodic：每隔一秒产生一个递增整数。
  //    它是无限流，永远不会自己结束，这里用 take(3) 截取前三个。
  final Stream<int> periodic = Stream<int>.periodic(
    const Duration(seconds: 1),
    (i) => i,
  ).take(3);

  // 2) Stream.fromFuture：把一个 Future 包装成 Stream
  final Future<int> future = Future.delayed(const Duration(seconds: 1), () => 1);
  final Stream<int> fromFuture = Stream<int>.fromFuture(future);

  // 3) Stream.fromFutures：把多个 Future 的结果依次推入同一个 Stream
  final Future<String> hello = Future.delayed(const Duration(seconds: 1), () => 'Hello');
  final Future<String> world = Future.delayed(const Duration(seconds: 2), () => 'World');
  final Stream<String> fromFutures = Stream.fromFutures([hello, world]);

  // 4) Stream.fromIterable + Stream.merge：把两个有限流合并成一条
  final Stream<int> left = Stream.fromIterable([1, 2, 3]);
  final Stream<int> right = Stream.fromIterable([4, 5, 6]);
  final Stream<int> merged = Stream.merge([left, right]);

  // await for 按到达顺序逐个取出合并后的数据并打印
  await for (final int i in merged) {
    print(i);
  }
}

为什么使用 await for？

await for 是 Dart 2.0 引入的异步迭代语法：它从 Stream 里逐个取数据，取到一次就执行一次循环体；没有数据时就挂起等待，数据到达才继续。它把「订阅 → 收数据 → 处理 → 完成」的整个过程，写成一段像普通 for 循环一样线性的代码。

把它和基于回调的 listen() 对比更好理解：

stream.listen(onData)：数据到达时触发回调，回调里再套回调，逻辑一多就不太好读。
await for：把同样的逻辑展开成顺序代码，读起来、改起来都更直观。底层依旧是非阻塞的，等待数据时不会卡住 UI 线程。

一个容易踩的坑：Stream 上的 error 事件不会被 await for 吞掉，而是作为异常重新抛出。需要处理错误时，用 try/catch 包住循环体捕获即可。

Stream 的基本使用流程

一个 Stream 的完整生命周期分三步：

创建 Stream：用 StreamController() 创建一个可控制的流，或用 Stream.periodic()、Stream.fromIterable() 等从已有数据直接构造。
订阅 Stream：用 Stream.listen() 监听数据；在 Flutter 里则常把流交给 StreamBuilder，让它在数据到达时自动重建 UI。
发布数据：通过 StreamController.sink.add(data) 往流里推数据，所有订阅者都会收到并作出响应。

综合示例：CounterBloc + StreamBuilder

// 数据源：对外暴露只读 stream，通过 sink 发布数据
class CounterBloc {
  final _controller = StreamController<int>();

  Stream<int> get stream => _controller.stream;

  void increment() => _controller.sink.add(1);

  void dispose() => _controller.close();
}

// 页面：用 StreamBuilder 订阅流，数据到达时自动重建
class CounterPage extends StatelessWidget {
  @override
  Widget build(BuildContext context) {
    final bloc = CounterBloc();

    return Scaffold(
      appBar: AppBar(title: const Text('Counter Page')),
      body: StreamBuilder<int>(
        stream: bloc.stream,
        builder: (context, snapshot) {
          return Text(
            snapshot.hasData ? '${snapshot.data}' : 'Waiting for data...',
          );
        },
      ),
      floatingActionButton: FloatingActionButton(
        onPressed: bloc.increment,
        tooltip: 'Increment',
        child: const Icon(Icons.add),
      ),
    );
  }
}

这个例子把前面的三步串了起来：CounterBloc 内部用一个私有的 StreamController<int> 作为数据源，对外暴露只读的 stream getter；increment() 通过 sink.add(1) 发布数据。CounterPage 用 StreamBuilder 订阅 bloc.stream——每当数据到达，builder 就会拿最新的 snapshot 重新构建 UI，计数立刻刷新；右下角的按钮点击后触发 increment()，完成一次「发布 → 订阅 → 更新」的闭环。

简化起见，示例把 CounterBloc 直接建在了 build() 里；真实项目中应由外层管理它的生命周期（例如放进 StatefulWidget 或依赖注入框架），并在页面销毁时调用 dispose() 关闭 StreamController，避免资源泄漏。

总结

Stream 是 Flutter 中非常重要的异步编程工具：生产者只管发布数据，消费者只关心订阅，两者互不感知，组件因此实现了解耦。利用这一特性，可以让代码更模块化、状态流转更可控——消息通知、收藏列表、表单校验这类「一处产生、多处响应」的场景，都是 Stream 的用武之地。

Stream 是什么？​

Stream 和 Future 的区别​

Stream 的分类​

Stream 的使用​

单订阅流的构造​

Stream 的基本使用流程​

综合示例：CounterBloc + StreamBuilder​

总结​