在 Dart 中创建流

作者：Lasse Nielsen
2013 年 4 月（2021 年 5 月更新）

dart:async 库包含两种对许多 Dart API 很重要的类型：Stream 和 Future。Future 表示单个计算的结果，而流是一系列结果。您监听流以获取结果（数据和错误）以及流关闭的通知。您还可以在监听时暂停或在流完成之前停止监听。

但本文不是关于使用流的。而是关于创建您自己的流。您可以通过几种方式创建流

• 转换现有流。
• 使用 async* 函数从头开始创建流。
• 使用 StreamController 创建流。

本文展示了每种方法的代码，并提供提示以帮助您正确实现流。

有关使用流的帮助，请参阅异步编程：流。

创建流的常见情况是您已经有一个流，并且您想基于原始流的事件创建一个新的流。例如，您可能有一个字节流，您想通过 UTF-8 解码输入将其转换为字符串流。最通用的方法是创建一个新的流，该流等待原始流上的事件，然后输出新的事件。示例

/// Splits a stream of consecutive strings into lines.
///
/// The input string is provided in smaller chunks through
/// the `source` stream.
Stream<String> lines(Stream<String> source) async* {
  // Stores any partial line from the previous chunk.
  var partial = '';
  // Wait until a new chunk is available, then process it.
  await for (final chunk in source) {
    var lines = chunk.split('\n');
    lines[0] = partial + lines[0]; // Prepend partial line.
    partial = lines.removeLast(); // Remove new partial line.
    for (final line in lines) {
      yield line; // Add lines to output stream.
    }
  }
  // Add final partial line to output stream, if any.
  if (partial.isNotEmpty) yield partial;
}

对于许多常见的转换，您可以使用 Stream 提供的转换方法，例如 map()、where()、expand() 和 take()。

例如，假设您有一个流 counterStream，它每秒发出一个递增的计数器。这是它可能的实现方式

var counterStream =
    Stream<int>.periodic(const Duration(seconds: 1), (x) => x).take(15);

要快速查看事件，您可以使用如下代码

counterStream.forEach(print); // Print an integer every second, 15 times.

要转换流事件，您可以在监听流之前调用该流上的转换方法（例如 map()）。该方法返回一个新的流。

// Double the integer in each event.
var doubleCounterStream = counterStream.map((int x) => x * 2);
doubleCounterStream.forEach(print);

除了 map() 之外，您还可以使用任何其他转换方法，例如以下方法

.where((int x) => x.isEven) // Retain only even integer events.
.expand((var x) => [x, x]) // Duplicate each event.
.take(5) // Stop after the first five events.

通常，转换方法是您所需要的全部。但是，如果您需要对转换进行更多控制，则可以使用 Stream 的 transform() 方法指定 StreamTransformer。平台库为许多常见任务提供了流转换器。例如，以下代码使用 dart:convert 库提供的 utf8.decoder 和 LineSplitter 转换器。

Stream<List<int>> content = File('someFile.txt').openRead();
List<String> lines = await content
    .transform(utf8.decoder)
    .transform(const LineSplitter())
    .toList();

创建新流的一种方法是使用异步生成器 (async*) 函数。当函数被调用时创建流，当流被监听时函数体开始运行。当函数返回时，流关闭。在函数返回之前，它可以使用 yield 或 yield* 语句在流上发出事件。

这是一个原始示例，它以固定的时间间隔发出数字

Stream<int> timedCounter(Duration interval, [int? maxCount]) async* {
  int i = 0;
  while (true) {
    await Future.delayed(interval);
    yield i++;
    if (i == maxCount) break;
  }
}

此函数返回一个 Stream。当监听该流时，主体开始运行。它会重复延迟请求的时间间隔，然后发出下一个数字。如果省略 maxCount 参数，则循环没有停止条件，因此流会永远输出越来越大的数字 - 或者直到监听器取消其订阅。

当监听器取消（通过调用 listen() 方法返回的 StreamSubscription 对象上的 cancel()）时，下次主体到达 yield 语句时，yield 将改为充当 return 语句。执行任何封闭的 finally 块，函数退出。如果函数尝试在退出之前生成一个值，则该操作将失败并充当返回。

当函数最终退出时，cancel() 方法返回的 Future 将完成。如果函数以错误退出，则 Future 将以该错误完成；否则，它将以 null 完成。

另一个更有用的例子是将一系列 Future 转换为流的函数

Stream<T> streamFromFutures<T>(Iterable<Future<T>> futures) async* {
  for (final future in futures) {
    var result = await future;
    yield result;
  }
}

此函数从 futures 可迭代对象请求一个新的 Future，等待该 Future，发出结果值，然后循环。如果 Future 以错误完成，则流将以该错误完成。

很少有 async* 函数从无到有地构建流。它需要从某个地方获取数据，而最常见的情况是从另一个流中获取数据。在某些情况下，例如上面的 Future 序列，数据来自其他异步事件源。但是，在许多情况下，async* 函数过于简单，无法轻松处理多个数据源。这就是 StreamController 类发挥作用的地方。

如果流的事件来自程序的不同部分，而不仅仅是来自可以通过 async 函数遍历的流或 Future，则使用 StreamController 创建和填充流。

StreamController 为您提供了一个新的流和一种在任何时候和从任何地方向流添加事件的方法。该流具有处理监听器和暂停所需的所有逻辑。您返回流并自己保留控制器。

以下示例（来自stream_controller_bad.dart）展示了 StreamController 的基本但有缺陷的用法，以实现先前示例中的 timedCounter() 函数。此代码创建要返回的流，然后基于定时器事件（既不是 Future 也不是流事件）将数据馈送到该流。

// NOTE: This implementation is FLAWED!
// It starts before it has subscribers, and it doesn't implement pause.
Stream<int> timedCounter(Duration interval, [int? maxCount]) {
  var controller = StreamController<int>();
  int counter = 0;
  void tick(Timer timer) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (maxCount != null && counter >= maxCount) {
      timer.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }

  Timer.periodic(interval, tick); // BAD: Starts before it has subscribers.
  return controller.stream;
}

与之前一样，您可以使用 timedCounter() 返回的流，如下所示

var counterStream = timedCounter(const Duration(seconds: 1), 15);
counterStream.listen(print); // Print an integer every second, 15 times.

timedCounter() 的这种实现存在一些问题

• 它在有订阅者之前开始生成事件。
• 即使订阅者请求暂停，它也会继续生成事件。

如下一节所示，您可以通过在创建 StreamController 时指定回调（例如 onListen 和 onPause）来解决这两个问题。

一般来说，流应该等待订阅者，然后才开始工作。async* 函数会自动执行此操作，但当使用 StreamController 时，您可以完全控制，甚至可以在不应该添加事件时添加事件。当流没有订阅者时，其 StreamController 会缓冲事件，如果流永远没有订阅者，则可能导致内存泄漏。

尝试将使用流的代码更改为以下内容

void listenAfterDelay() async {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  await Future.delayed(const Duration(seconds: 5));

  // After 5 seconds, add a listener.
  await for (final n in counterStream) {
    print(n); // Print an integer every second, 15 times.
  }
}

当此代码运行时，前 5 秒没有任何输出，尽管流正在工作。然后添加监听器，并且前 5 个左右的事件会一次性打印出来，因为它们被 StreamController 缓冲了。

要获得订阅通知，请在创建 StreamController 时指定一个 onListen 参数。当流获得其第一个订阅者时，将调用 onListen 回调。如果您指定 onCancel 回调，则当控制器失去其最后一个订阅者时，将调用该回调。在前面的示例中，Timer.periodic() 应移动到 onListen 处理程序，如下一节所示。

避免在监听器请求暂停时生成事件。当流订阅暂停时，async* 函数会在 yield 语句处自动暂停。另一方面，StreamController 会在暂停期间缓冲事件。如果提供事件的代码不尊重暂停，则缓冲区的大小可能会无限增长。此外，如果监听器在暂停后很快停止监听，则创建缓冲区所花费的工作就会被浪费。

要了解在没有暂停支持的情况下会发生什么，请尝试将使用流的代码更改为以下内容：

void listenWithPause() {
  var counterStream = timedCounter(const Duration(seconds: 1), 15);
  late StreamSubscription<int> subscription;

  subscription = counterStream.listen((int counter) {
    print(counter); // Print an integer every second.
    if (counter == 5) {
      // After 5 ticks, pause for five seconds, then resume.
      subscription.pause(Future.delayed(const Duration(seconds: 5)));
    }
  });
}

当暂停的五秒钟结束后，这段时间内触发的所有事件会立即被接收。发生这种情况是因为流的源不响应暂停，并持续向流添加事件。因此，流会缓冲这些事件，然后在流取消暂停时清空其缓冲区。

以下版本的 timedCounter()（来自 stream_controller.dart）通过使用 StreamController 上的 onListen、onPause、onResume 和 onCancel 回调来实现暂停。

Stream<int> timedCounter(Duration interval, [int? maxCount]) {
  late StreamController<int> controller;
  Timer? timer;
  int counter = 0;

  void tick(_) {
    counter++;
    controller.add(counter); // Ask stream to send counter values as event.
    if (counter == maxCount) {
      timer?.cancel();
      controller.close(); // Ask stream to shut down and tell listeners.
    }
  }

  void startTimer() {
    timer = Timer.periodic(interval, tick);
  }

  void stopTimer() {
    timer?.cancel();
    timer = null;
  }

  controller = StreamController<int>(
      onListen: startTimer,
      onPause: stopTimer,
      onResume: startTimer,
      onCancel: stopTimer);

  return controller.stream;
}

使用上面的 listenWithPause() 函数运行此代码。你会看到它在暂停时停止计数，并在之后顺利恢复。

你必须使用所有侦听器——onListen、onCancel、onPause 和 onResume——才能收到暂停状态更改的通知。原因是，如果订阅和暂停状态同时发生更改，则只会调用 onListen 或 onCancel 回调。

在不使用 async* 函数创建流时，请记住以下提示：

• 使用同步控制器时要小心，例如，使用 StreamController(sync: true) 创建的控制器。当你向未暂停的同步控制器发送事件时（例如，使用 EventSink 定义的 add()、addError() 或 close() 方法），该事件会立即发送到流上的所有侦听器。在添加侦听器的代码完全返回之前，绝不能调用 Stream 侦听器，而在错误的时间使用同步控制器可能会打破这个承诺，并导致良好的代码失败。避免使用同步控制器。

• 如果你使用 StreamController，则 onListen 回调会在 listen 调用返回 StreamSubscription 之前被调用。不要让 onListen 回调依赖于订阅已经存在。例如，在以下代码中，onListen 事件会在 subscription 变量具有有效值之前触发（并调用 handler）。

  dart

  subscription = stream.listen(handler);

• StreamController 定义的 onListen、onPause、onResume 和 onCancel 回调会在流的侦听器状态更改时被流调用，但永远不会在事件触发期间或在调用另一个状态更改处理程序期间被调用。在这些情况下，状态更改回调会延迟到前一个回调完成后才会执行。

• 不要尝试自己实现 Stream 接口。很容易在事件、回调以及添加和删除侦听器之间的交互中出现微妙的错误。始终使用现有的流，可能来自 StreamController，来实现新流的 listen 调用。

• 虽然可以通过扩展 Stream 类并实现 listen 方法和额外的功能来创建扩展 Stream 并具有更多功能的类，但这通常不建议这样做，因为它引入了用户必须考虑的新类型。与其创建一个是 Stream（并且更多）的类，不如创建一个拥有 Stream（并且更多）的类。

除非另有说明，本站上的文档反映的是 Dart 3.6.0。页面最后更新于 2021-05-16。 查看源代码 或者 报告问题。