Home未分类(译)异步编程: Streams

(译)异步编程: Streams

2011 年 4 月 10 日

原文链接

知识点有哪些？

listen()

Dart中的异步编程以 Future
和 Stream
类为特征。
Future表示不立即完成的计算。在普通函数返回结果的地方，异步函数返回一个 Future ，最终将包含结果。Future 将告诉你何时结果准备就绪。
流是一系列异步事件。它就像一个异步的Iterable，流在准备好事件的时候告诉你,而不是在你要求它时获得下一个事件。

接收流事件

流可以通过多种方式创建，这是另一篇文章的主题，但它们都可以以相同的方式使用：异步 for循环(通常称为 await for）迭代流的事件就像 for 循环迭代在一个 Iterable。例如：

Future sumStream(Stream stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

此代码只接收整数事件流的每个事件，将它们相加，然后返回总和。循环体结束时，函数暂停，直到下一个事件到达或流完成。

该函数使用 async
关键字标记，这在使用 await for 循环时是必需的。
以下示例通过使用async* 函数生成简单的整数流来测试前面的代码：

// Copyright (c) 2015, the Dart project authors.
// Please see the AUTHORS file for details.
// All rights reserved. Use of this source code is governed
// by a BSD-style license that can be found in the LICENSE file.

import 'dart:async';

Future sumStream(Stream stream) async {
  var sum = 0;
  await for (var value in stream) {
    sum += value;
  }
  return sum;
}

Stream countStream(int to) async* {
  for (int i = 1; i <= to; i++) {
    yield i;
  }
}

main() async {
  var stream = countStream(10);
  var sum = await sumStream(stream);
  print(sum); // 55
}

错误事件

当流中没有更多事件时流就结束了，并且就像通知新事件到达一样接收事件的代码会被通知。使用 await for 循环读取事件时，循环在流完成时停止。
在某些情况下，在流完成之前会发生错误;也许从远程服务器获取文件时网络失败，或者创建事件的代码可能存在错误，但有人需要了解它。
Streams 还可以提供错误事件，例如它提供数据事件。大多数流将在第一个错误之后停止，但是可能有多个流提供多个错误，并且流可以在错误事件之后提供更多数据。在本文档中，我们仅讨论最多传递一个错误的流。
使用 await for 读取流时，循环语句抛出错误。这也结束了循环。你可以使用 try-catch 捕获错误。以下示例在循环迭代器等于4时抛出错误：

使用 streams

Stream 类包含许多辅助方法，可以为你在流上执行常见操作，类似于Iterable上的方法。例如，你可以使用Stream API中的lastWhere()在流中找到最后一个正整数。

Future lastPositive(Stream stream) =>
    stream.lastWhere((x) => x >= 0);

两种类型的流

存在两种类型的流。

单个订阅流

最常见的流包含一系列事件，这些事件是更大整体的一部分。事件需要以正确的顺序传递，而不会遗漏任何事件。这是你在读取文件或接收Web请求时获得的流。
这样的流只能被听一次。稍后再次倾听可能意味着错过了初始事件，然后流的其余部分毫无意义。当你开始收听时，将获取数据并以块的形式提供。

广播流

另一种流用于可以一次处理一个的单个消息。例如，这种流可以用于浏览器中的鼠标事件。
您可以随时开始监听此类流，然后你可以获得在监听时触发的事件。多个监听者可以同时监听，您可以在取消之前的订阅后再次监听。

处理流的方法

Stream上的以下方法处理流并返回结果：

Future get first;
Future get isEmpty;
Future get last;
Future get length;
Future get single;
Future any(bool Function(T element) test);
Future contains(Object needle);
Future drain([E futureValue]);
Future elementAt(int index);
Future every(bool Function(T element) test);
Future firstWhere(bool Function(T element) test, {T Function() orElse});
Future fold(S initialValue, S Function(S previous, T element) combine);
Future forEach(void Function(T element) action);
Future join([String separator = ""]);
Future lastWhere(bool Function(T element) test, {T Function() orElse});
Future pipe(StreamConsumer streamConsumer);
Future reduce(T Function(T previous, T element) combine);
Future singleWhere(bool Function(T element) test, {T Function() orElse});
Future<List> toList();
Future<Set> toSet();

除了 drain() 和 pipe() 之外的所有这些函数都对应于 Iterable 上的类似函数。通过使用具有等待循环的异步函数（或仅使用其他方法之一）,可以轻松地编写每个函数。例如，一些实现可能是：

Future contains(Object needle) async {
  await for (var event in this) {
    if (event == needle) return true;
  }
  return false;
}

Future forEach(void Function(T element) action) async {
  await for (var event in this) {
    action(event);
  }
}

Future<List> toList() async {
  final result = [];
  await this.forEach(result.add);
  return result;
}

Future join([String separator = ""]) async =>
    (await this.toList()).join(separator);

（实际实现稍微复杂一些，但主要是出于历史原因。）

修改流的方法

Stream上的以下方法基于原始流返回新流。每个人都等待，直到有人在监听原始流之前监听新流。

Stream cast();
Stream expand(Iterable Function(T element) convert);
Stream map(S Function(T event) convert);
Stream retype();
Stream skip(int count);
Stream skipWhile(bool Function(T element) test);
Stream take(int count);
Stream takeWhile(bool Function(T element) test);
Stream where(bool Function(T event) test);

前面的方法对应于 Iterable 上的类似方法，它将 iterable 转换为另一个 iterable 。所有这些都可以使用带有等待循环的异步函数轻松编写。

Stream asyncExpand(Stream Function(T event) convert);
Stream asyncMap(FutureOr Function(T event) convert);
Stream distinct([bool Function(T previous, T next) equals]);

asyncExpand() 和asyncMap() 函数类似于 expand() 和 map() ，但允许其函数参数为异步函数。 distinct() 函数在 Iterable 上不存在，但它可能有。

Stream handleError(Function onError, {bool test(error)});
Stream timeout(Duration timeLimit,
    {void Function(EventSink sink) onTimeout});
Stream transform(StreamTransformer streamTransformer);

最后三个功能更加特殊。它们涉及错误处理，等待循环无法做到 – 到达循环的第一个错误将结束循环及其在流上的订阅。没有恢复。在 await for 循环中使用它之前，您可以使用handleError() 从流中删除错误。

transform()方法

transform() 函数不仅用于错误处理;它是流的更通用的 “map”。通常的map要求每个传入事件都有一个值。但是，特别是对于 I/O 流，可能需要多个传入事件才能生成输出事件。 StreamTransformer
可以使用它。例如，像 Utf8Decoder
这样的解码器就是变换器。变换器只需要一个函数 bind()
，它可以通过异步函数轻松实现。 

Stream mapLogErrors(
  Stream stream,
  S Function(T event) convert,
) async* {
  var streamWithoutErrors = stream.handleError((e) => log(e));
  await for (var event in streamWithoutErrors) {
    yield convert(event);
  }
}

读取和解析一个文件

以下代码读取文件并在流上运行两个转换。它首先从UTF8转换数据，然后通过LineSplitter运行它。打印所有行，除了以＃标签开头的任何行＃。

import 'dart:convert';
import 'dart:io';

Future main(List args) async {
  var file = File(args[0]);
  var lines = file
      .openRead()
      .transform(utf8.decoder)
      .transform(LineSplitter());
  await for (var line in lines) {
    if (!line.startsWith('#')) print(line);
  }
}

监听方法

Stream上的 final 方法是 listen()。这是一种“低层”方法 – 所有其他流函数都是根据listen()定义的。

StreamSubscription listen(void Function(T event) onData,
    {Function onError, void Function() onDone, bool cancelOnError});

要创建新的 Stream 类型，您可以扩展 Stream 类并实现 listen() 方法 – Stream 调用 listen() 上的所有其他方法以便工作。

listen()方法允许你开始监听流。在你这样做之前，流是一个惰性对象，描述您想要查看的事件。监听时，将返回 StreamSubscription
对象，该对象表示生成事件的活动流。这类似于 Iterable
只是一个对象集合，但迭代器是进行实际迭代的迭代器。
流订阅允许你暂停订阅，暂停后恢复它，并完全取消订阅。你可以为每个数据事件或错误事件以及关闭流时设置回调。

Related Posts

About The Author

maynard

懒散，不想无聊