ShowDocs
搜索文章Ctrl K

外观

← 返回Dart 与 Flutter列表
知识专题

Dart Future、Stream 与取消边界 ​

前置：类型与函数，尤其要理解回调和Iterable。异步返回值不是“后台线程”，并发、并行、取消是三个需要分别定义的能力。

Future、await 与 timeout ​

Future<T>表示一次将来的成功值或错误。async函数遇到未完成await后暂停自身，把后续执行交给事件循环；同一isolate内一次只执行一段Dart代码。调用async函数会开始执行，等它返回的Future不意味着自动再启动一份工作。try { await work; } catch (...)能观察异步失败；只在同步调用外层try而不await，通常捕获不到后续错误。每个启动的Future都应由某一层等待或处理错误，不能让未观察错误变成进程级意外。

Future没有通用cancel方法。future.timeout(duration)返回另一个受期限约束的Future，原操作可继续跑、继续占资源或产生副作用。对文件/网络的取消必须寻找底层提供的协议；若自己启动Timer则在退出路径cancel，若有控制器则关闭并等待清理。不能把“等候者不再等”写成“工作已停止”。完整test.dart明确让原Future在timeout之后仍完成，并检查finished=true。

事件队列与微任务负责安排延续；大量递归微任务可能让普通事件难以执行。不要用Future或scheduleMicrotask包装CPU死循环期望界面恢复；必须减少单次工作、分批让出或移到isolate。Future.wait并不会为每项分配一个CPU核，也不会让所有底层操作在某个错误后自动取消。需要控制并发度时明确限制活跃工作数量，并在退出前观察所有已启动任务。

Stream、订阅、错误与资源 ​

Stream<T>表达多次事件。单订阅流通常适合文件顺序读取；广播流允许多个监听者，但没有订阅时事件可能丢失，不能把广播当持久化队列。await-for每次获取一项并等待循环体完成，适合将文件字节流经UTF-8解码和LineSplitter转为行。链路的缓冲和pause行为依赖生产者，不能声称await-for对所有来源都提供固定内存背压。

结束循环、处理抛错或显式调用await subscription.cancel()会取消对应订阅；cancel返回Future，应该等其清理完成。订阅取消与原Future超时不同：前者有明确生产者通知接口，但生产者是否停止外部工作仍由实现决定。库接收外部Stream时要写清消费会占用单次订阅，以及错误时是否退出。我们的summarize借用行流、消费到结束或首个坏行；抛错会让await-for取消上游订阅。test.dart用StreamController的onCancel证明发生取消，不只检查错误文字。

File.openRead创建的文件流在结束/取消时管理文件句柄；若自己创建RandomAccessFile则需try/finally关闭，不要期待GC及时释放。某些sink有close/done两个阶段：停止写入后等待真正完成，写入错误也需观察。不要在输出还没完成时直接exit丢掉缓冲；本项目设置exitCode让运行时完成退出。

isolate、消息与可重复性能实验 ​

isolate有独立状态和事件循环，通常通过消息传递而不是共享可变列表通信。CPU密集计算可以考虑Isolate.run；创建、复制或传递大数据有成本，不是每行账目起一个isolate。取消长任务需要自己设计请求/确认协议或明确终止isolate的清理边界；强制终止不等价于finally已完成。主isolate退出前应等待子工作结果或取消确认。

下面完整Lab同时检查惰性Iterable、不可修改快照、单订阅first的取消和相同数据的惰性/物化求和：

lab.dart
dart
import 'dart:async';
Future<void> main() async {
  var visits = 0;
  final source = [1, 2, 3];
  final lazy = source.where((n) { visits++; return n > 1; });
  source.add(4);
  if (lazy.length != 3 || lazy.length != 3 || visits != 8) throw StateError('lazy contract');
  final snapshot = List<int>.unmodifiable(lazy);
  source.add(5);
  if (snapshot.length != 3 || lazy.length != 4) throw StateError('snapshot contract');
  var stopped = false;
  final controller = StreamController<int>(onCancel: () { stopped = true; });
  final first = controller.stream.first;
  controller.add(42);
  if (await first != 42 || !stopped) throw StateError('subscription not cancelled');
  await controller.close();
  for (final materialize in [false, true]) {
    final watch = Stopwatch()..start();
    final values = Iterable<int>.generate(100000).where((n) => n.isEven).map((n) => n * n);
    final sum = (materialize ? values.toList() : values).fold<int>(0, (a, b) => a + b);
    watch.stop();
    if (sum != 166661666700000) throw StateError('unequal semantics');
    print('materialize=$materialize sum=$sum micros=${watch.elapsedMicroseconds}');
  }
  print('lazy collection and single-subscription cancellation passed');
}
shell
dart run lab.dart

两次sum均为166661666700000；计时行materialize=false/true的micros因JIT、分配与运行顺序而变，不要求固定比例。一次运行不能证明哪一种普遍更快。重复测量可交替顺序、预热、保持相同输入，并用DevTools的CPU采样与内存视图定位分配热点；dart --observe lab.dart适合让VM暴露服务以便连接，短程序可能很快退出，需要调试时在断点暂停。不要把微秒变化误认为泄漏证据，泄漏关注GC后仍被引用的对象。

练习与参考解法 ​

  1. 给一个Future加Duration.zero超时，然后继续等待原Future，预测副作用是否执行。2. 为什么broadcast流不适合重新读取同一个文件？3. 在文件格式错误后如何证明上游取消？4. 循环计算一亿次加法加async是否释放CPU？
参考解法

超时只结束包装Future的等待，原工作仍可能完成；测试必须等它再检查副作用。广播不是可重放文件缓冲，后订阅者可能漏事件；重新打开文件或显式缓存要由业务决定。使用可控制StreamController的onCancel，在收到坏行后等待summarize失败并确认回调，再close控制器。async不自动让CPU计算变并行；分块、isolate或更有效算法需根据任务选择。

依据：英文Streams、中文async/await、Future.timeout API、并发与isolate。

继续阅读