内容

在 Dart 中创建流

内容

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

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

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

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

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

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

转换现有流

#

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

dart
/// 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，每秒钟发出一个递增的计数器。以下是它可能实现的方式

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

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

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

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

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

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

dart
.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。平台库为许多常见任务提供了流转换器。例如，以下代码使用 utf8.decoder 和 LineSplitter 转换器（由 dart:convert 库提供）。

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

从头开始创建流

#

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

以下是一个简单的示例，它以规则的间隔发出数字

dart
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 转换为流的函数

dart
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 类发挥作用的地方。

使用 StreamController

#

如果您的流的事件来自程序的不同部分，而不仅仅来自可以由 async 函数遍历的流或 future，那么使用 StreamController 来创建和填充流。

StreamController 为您提供了一个新流，以及一种在任何时间点和任何位置向流添加事件的方法。流具有处理监听器和暂停所需的所有逻辑。您返回流，并将控制器保留给自己。

以下示例（来自 stream_controller_bad.dart）展示了 StreamController 的基本用法（尽管存在缺陷），用于实现前面示例中的 timedCounter() 函数。此代码创建一个要返回的流，然后根据计时器事件（既不是 future 也不是流事件）将数据馈送到流中。

baddart
// 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() 返回的流

dart
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 会缓冲事件，如果流永远没有订阅者，这会导致内存泄漏。

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

dart
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 会在暂停期间缓冲事件。如果提供事件的代码不尊重暂停，则缓冲区的大小可能会无限增长。此外，如果监听器在暂停后不久停止监听，那么用于创建缓冲区的资源就会浪费。

要查看没有暂停支持的情况，请尝试将使用流的代码更改为以下内容

dart
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)));
    }
  });
}

当 5 秒的暂停时间到期后，在此期间触发的事件会立即全部接收。这是因为流的源代码不尊重暂停，并继续将事件添加到流中。因此，流会缓冲事件，并在流恢复暂停后清空其缓冲区。

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

dart
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) 创建的控制器。当您在未暂停的同步控制器上发送事件时（例如，使用 add()、addError() 或 close() 方法，这些方法由 EventSink 定义），该事件会立即发送到流上的所有监听器。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.5.3。页面最后更新于 2021-05-16。 查看源代码 或 报告问题.