理解 Dart 异步事件流 Stream

hanguokai

原文：http://han.guokai.blog.163.com/b ... 271201323083652728/

基本概念

顾名思义，Stream 就是流的意思，表示发出的一系列的异步数据。可以简单地认为 Stream 是一个异步数据源。它是 Dart 中处理异步事件流的统一 API。

集合与Stream
Dart 中，集合（Iterable或Collection）表示一系列的对象。而 Stream （也就是“流”）也表示一系列的对象，但区别在于 Stream 是异步的事件流。比如文件、套接字这种 IO 数据的非阻塞输入流（input data），或者用户界面上用户触发的动作（UI事件）。

集合可以理解为“拉”模式，比如你有一个 List ，你可以主动地通过迭代获得其中的每个元素，想要就能拿出来。而 Stream 可以理解为“推”模式，这些异步产生的事件或数据会推送给你（并不是你想要就能立刻拿到）。这种模式下，你要做的是用一个 listener （即callback）做好数据接收的准备，数据可用时就通知你。

推和拉就是别人给你还是你自己去拿的区别。但是不管如何获取数据，二者的本质都可以认为是数据的集合（数据可能无限多）。所以，二者有很多相同的方法，稍后介绍。

怎么理解 Stream 中的数据？

数据（data）是个非常抽象的概念，可以认为一切皆数据。在程序的世界里，其实只有两种东西：数据和对数据的操作。对数据的操作就是对输入的数据经过一些计算，之后输出一些新数据。事件（event，如UI上的事件）、计算结果（value，如函数/方法的返回值）以及从文件或网络获得的纯数据都可以认为是数据（data）。另外，Dart 中的所有事物都是对象，所以数据也一定是某种对象（object）。在本文中，可以认为事件、结果、数据、对象都是一样的，不用特意区分。

Stream 与 Future

Stream 和 Future 是 Dart 异步处理的核心 API。Future<T> 表示稍后获得的一个数据，所有异步的操作的返回值都用 Future<T> 来表示。但是 Future 只能表示一次异步获得的数据。而 Stream<T> 表示多次异步获得的数据。比如界面上的按钮可能会被用户点击多次，所以按钮上的点击事件（onClick）就是一个 Stream<MouseEvent> 。简单地说，Future将返回一个值，而Stream将返回多次值。

另外一点， Stream 是流式处理，比如 IO 处理的时候，一般情况是每次只会读取一部分数据（具体取决于实现）。和一次性读取整个文件的内容相比，Stream 的好处是处理过程中内存占用较小。而 File 的 readAsString（异步读，返回 Future<String>）或 readAsStringSync（同步读，返回 String）等方法都是一次性读取整个文件的内容进来，虽然获得完整内容处理起来比较方便，但是如果文件很大的话就会导致内存占用过大的问题。

基本使用

获得 Stream
Dart 中统一使用 Stream 处理异步事件流，所以可以获得 Stream 的地方很多。为了方便演示，这里先介绍2种获取 Stream 的方式。

1. 将集合（Iterable）包装为 Stream
Stream 有3个工厂构造函数：fromFuture、fromIterable 和 periodic，分别可以通过一个 Future、Iterable或定时触发动作作为 Stream 的事件源构造 Stream。下面的代码就是通过一个 List<int> 构造的 Stream<int>。

1.   var data = [1, 2, 3, 4];
   
2.   var stream = new Stream.fromIterable(data);

复制代码

对集合的包装只是简单地模拟异步，定时触发、IO输入、UI事件等现实情况才是真正的异步事件。

2. 使用 Stream 读文件
读文件的方式有多种，其中一种是使用 Stream 获得文件内容。File 的方法 openRead() 返回一个 Stream<List<int>>，List<int> 可以理解为一个 byte array，因为 Dart 中没有 byte 类型。下面的代码将打开当前程序的源代码的 Stream 输入流。

1.   var stream = new File(new Options().script).openRead();

复制代码

订阅 Stream

当你有了一个 Stream 时，最常用的功能就是通过 listen() 方法订阅 Stream 上发出的数据（即事件）。有事件发出时就会通知订阅者。如果在发出事件的同时添加订阅者，那么要在订阅者在该事件发出后才会生效。如果订阅者取消了订阅，那么它会立即停止接收事件。


我们在接收一个输入流的时候要面临几种不同的情况和状态，最基本的是处理收到数据，此外上游还可能出现错误，以及出现错误时是否继续后续数据的处理，最后在输入完成的时候还有一个结束状态。所以 listen 方法的几个参数分别对应这些情况和状态：

• onData，处理收到的数据的 callback
• onError，处理遇到错误时的 callback
• onDone，结束时的通知 callback
• unsubscribeOnError，遇到第一个错误时是否停止（也就是取消订阅），默认为false
  

onData 是唯一必填参数，也是用的最多的，后面3个是可选的命名参数。


下面我们订阅一个 Stream 的数据，收到数据时只是简单地打印出来：

1.   var data = [1, 2, 3, 4];
   
2.   var stream = new Stream.fromIterable(data);
   
3.   
   
4.   stream.listen((e)=>print(e), onDone: () => print('Done'));
   
5.   // => 1, 2, 3, 4
   
6.   // => Done

复制代码

上面的代码会先打印出从 Stream 收到的每个数字，最后打印一个‘Done'。

当 Stream 中的所有数据发送完时，就会触发 onDone 的调用，但提前取消订阅不会触发 onDone 。在结束的同时（收到 onDone 事件之前），所有的订阅者都被取消了订阅，此时 Stream 上便没有订阅者了。允许对一个已经结束了的 Stream 再添加订阅者（尽管没什么意义），此时只会立刻收到一个 onDone 事件。

1.   stream.listen(print, onDone: () {
   
2.     print('first done');
   
3.     //listen again
   
4.     stream.listen(print, onDone:() => print('second done'));
   
5.   });
   
6.   // => data: 1,2,3,4,
   
7.   // => first done
   
8.   // => no data, because stream is done
   
9.   // => second done

复制代码

上面的代码中，首先我们在 onDone 的回调中打印了 'first done' 表示第一次结束。此时 stream 上已经没有订阅者了，但接着我们又再次订阅了这个 stream。这一次没有再收到数据，而是马上打印出了 ‘second done’ 表示第二次订阅的结束。

高级订阅管理
前面的示例代码会处理 Stream 发出的所有数据，直到 Stream 结束。如果想提前取消处理怎么办？listen() 方法会返回一个 StreamSubscription 对象，用于提供对订阅的管理控制。onData、onError和onDone 这3个方法分别用于设置（如果listen方法中的参数为null）或覆盖对应的 callback。cancel、pause和resume分别用于取消订阅、暂停和继续。比如，可以在 listen 方法中参数置为 null，接着通过 subscription 对象设置 callback 。此外，cancel 方法也重要，要么一直处理数据直到 stream 结束，要么提前取消订阅结束处理。比如使用 Stream 读文件，为了使资源得到释放，要么读完整个文件，要么使用 subscription 的 cancel 方法取消订阅（即终止后续数据的读取）。可以看出，这里的 cancel 相当于传统意义上的 close 方法。最后，pause和resume方法是尝试向数据源发出暂停和继续的请求，其意义取决于实际情况，并且不保证一定能生效。比如数据源能够支持，或者是带缓冲实现的 stream 才能做到暂停。

1.   var sub = stream.listen(null);
   
2.   sub.onData(print);
   
3.   sub.onError((e)=>print('error $e'));
   
4.   sub.onDone(()=>print('done'));
   
5.   // => 1, 2, 3, 4, done

复制代码

上面的代码与前面的 listen 示例代码作用相同。

1.   var sub = stream.listen(null);
   
2.   sub.onData((e){
   
3.     if(e > 2)
   
4.       sub.cancel();
   
5.     else
   
6.       print(e);
   
7.   });
   
8.   sub.onDone(()=>print('done'));
   
9.   // => 1, 2
   
10.   // no 'done', because stream is cancel.

复制代码

上面的代码最后会打印出1和2，但不会打印出‘done' 。首先，listen 中的参数为 null，也就是没有订阅者。然后，通过 listen 的返回者 subscription 对象设置了 onData 和 onDone 的处理，这时才有了订阅者。在 onData 中，如果收到的数字大于2就取消后续处理，因此到数字 3 的时候就没有打印 3，而是立即结束了处理，这样后面的 4 也不会出现了。既然是提前退出，所以 onDone 也是不会触发的。

Stream 两种订阅模式
Stream有两种订阅模式：单订阅(single)和多订阅（broadcast）。单订阅就是只能有一个订阅者，而广播是可以有多个订阅者。这就有点类似于消息服务（Message Service）的处理模式。单订阅类似于点对点，在订阅者出现之前会持有数据，在订阅者出现之后就才转交给它。而广播类似于发布订阅模式，可以同时有多个订阅者，当有数据时就会传递给所有的订阅者，而不管当前是否已有订阅者存在。

Stream 默认处于单订阅模式，所以同一个 stream 上的 listen 和其它大多数方法只能调用一次，调用第二次就会报错。但 Stream 可以通过 transform() 方法（返回另一个 Stream）进行连续调用。通过 Stream.asBroadcastStream() 可以将一个单订阅模式的 Stream 转换成一个多订阅模式的 Stream，isBroadcast 属性可以判断当前 Stream 所处的模式。

1.   assert(stream.isBroadcast == false);
   
2.   stream.first.then(print);
   
3.   stream.last.then(print);// Bad state: Stream already has subscriber.

复制代码

上的代码需要分别打印出 stream 的第一个数据和最后一个数据，但是单模式 Stream 只能订阅一次，所以直接出错了。当然，Stream 是异步的，所以 first 也没有打印出来。

1.   var bs = stream.asBroadcastStream();
   
2.   assert(bs.isBroadcast == true);
   
3.   bs.first.then(print);
   
4.   bs.last.then(print);
   
5.   // OK => 1, 4

复制代码

上面的代码，我们把单模式 Stream 转成了多订阅的 Stream，所以可以 first 和 last 都打印出来了。

按前面说的，单订阅模式会持有数据，多订阅模式如果没有及时添加订阅者则可能丢数据。不过具体取决于 stream 的实现。

1.   new Timer(new Duration(seconds:5), ()=>stream.listen(print));
   
2.   // after 5 second, it output 1,2,3,4

复制代码

上面的代码利用 Timer 延迟了5秒才订阅 stream，但仍然输出了数据。因为我们这里的这个 stream 是单订阅模式，它在有订阅者后才会发出事件。那么多订阅模式就一定会漏掉数据吗？

1.   var bs = stream.asBroadcastStream();
   
2.   new Timer(new Duration(seconds:5), ()=>bs.listen(print));
   
3.   // after 5 second, it also output 1,2,3,4
   
4.   // because asBroadcastStream() is a simple wrap,
   
5.   // it don't change the source stream's feature

复制代码

上面我们把原始的单订阅模式转成了多订阅模式的 Stream，此时可以添加多个订阅者。我们5秒后才在 broadcast stream 上添加了订阅者，但它依然输出了 1,2,3,4 ，并没有漏掉数据。这其实是因为 asBroadcastStream() 只是对原始 stream 的封装，并不改变原始 stream 的实现特性。所以这个 broadcast stream 同样在等待有订阅者之后才发出数据。但是如果一旦有了第一个订阅者，然后再延迟添加第二个订阅者就会漏数据了。

1.   var bs = stream.asBroadcastStream();
   
2.   // add first listener
   
3.   new Timer(new Duration(seconds:5), ()=>bs.listen(print));
   
4.   // after 5 second, it output 1,2,3,4
   
5. 
   
6.   // add second listener
   
7.   new Timer(new Duration(seconds:10), ()=>bs.listen(print));
   
8.   // after 10 second, nothing output, because stream is done

复制代码

再来看另外一个例子，我们自己来创建一个 Stream。StreamController 用于创建 Stream，它有两个构造函数，分别用于创建单订阅模式 Stream 和 多订阅模式 Stream。然后可以利用 add()、addError() 和 close() 方法发送事件、发送错误和结束，这三个方法来自 EventSink，是各种 Sink 上的通用方法。

1.   // build single stream
   
2.   //var controller = new StreamController();
   
3. 
   
4.   // build broadcast stream
   
5.   var controller = new StreamController.broadcast();
   
6.   //send event
   
7.   controller..add(1)
   
8.                 ..add(2)
   
9.                 ..add(3)
   
10.                 ..add(4);
    
11.   //send done
    
12.   controller.close();
    
13. 
    
14.   var myStream = controller.stream;
    
15.   new Timer(new Duration(seconds:5), ()=>myStream.listen(print));
    
16.   //if myStream is single stream, it output 1,2,3,4
    
17.   //if myStream is broadcast stream, it output nothing, because stream is done.

复制代码

Stream 的集合特性
前面说过，Stream 和一般的集合类似，都是一组数据，只不过一个是异步推送，一个是同步拉取。所以他们都很多共同的方法。例如：

1.   stream.any((e) => e > 2).then(print);// stream.any()
   
2.   print([1,2,3,4].any((e) => e > 2));// iterable.any()
   
3.   // => true, true

复制代码

比如 Stream 和 集合 都有 any() 方法，集合是同步的（但是惰性执行，这里因为有 print 调用，所以立刻执行了）并直接返回结果， Stream 上的 any() 方法是异步的，返回的是 Future 。方法本身的含义都是一样的。上面的代码虽然 stream 的 any 方法在前，但因为是异步的，所以的输出在后。

在列举其它 Stream 和 Iterable 通用的方法：

1.   //常见集合方法
   
2.   stream.first.then(print);
   
3.   stream.firstWhere((e)=>e>3, defaultValue:()=>0).then(print);
   
4.   stream.last.then(print);
   
5.   stream.lastWhere((e)=>e>3, defaultValue:()=>0).then(print);
   
6.   stream.length.then(print);
   
7.   stream.isEmpty.then(print);
   
8.   
   
9.   stream.any((e) => e > 2).then(print);
   
10.   stream.every((e) => e > 2).then(print);
    
11.   stream.contains(3).then(print);
    
12.   stream.elementAt(2).then(print);
    
13.   stream.where((e) => e >2).listen(print);
    
14. 
    
15.   stream.skip(2).listen(print);
    
16.   stream.skipWhile((e) => e < 2).listen(print);
    
17.   stream.take(2).listen(print);
    
18.   stream.takeWhile((e)=>e<3).listen(print);
    
19. 
    
20.   stream.map((e) => e*2).listen(print);
    
21.   stream.reduce(0, (p, c) => p + c).then(print);
    
22.   stream.expand((e) => [e, e]).listen(print);
    
23.   
    
24.   stream.toList().then(print);
    
25.   stream.toSet().then(print);

复制代码

注意以上方法同时只能使用一次，因为是单订阅模式。此外，如果方法只有一个返回值，即数据收敛类型的方法，那么返回就是一个 Future。如果是只是数据转换的方法，如 map ，返回的还是一个 Stream，只是数据数据的类型和数量变了。看到这么多 Stream 与 Iterable 相同的方法，大家应该更清楚 Stream 其实也是个数据集合。

通用数据收敛方法
集合中有很多方法只返回一个值，多个数据作为输入、一个数据作为输出的方法就是数据收敛的方法。Stream 有一个更通用的收敛方法 pipe() 。pipe() 方法的参数要求是一个 StreamConsumer 接口的实现，该接口只有一个方法： Future<T> consume(Stream<S> stream)

1.   class DataConsumer implements StreamConsumer{
   
2.     Future consume(Stream stream){
   
3.       return stream.reduce(0, (c,p)=>c+p);
   
4.     }
   
5.   }
   
6. 
   
7.   stream.pipe(new DataConsumer()).then(print);
   
8.   // => 10
   
9. 
   
10.   // equivalent below
    
11.   stream.reduce(0, (p, c) => p + c).then(print);

复制代码

上面我们自己实现了一个 StreamConsumer ，它只是对 Stream 的数据求和，并返回该结果。这个简单的例子实际意义不大。但这里只是为了演示这个通用 pipe() 方法和 StreamConsumer 接口的意义。

通用数据转换方法
除了数据收敛方法，Stream 也有自己通用的数据转换方法 transform() 。类似于 Future 的连续调用，Stream 也可以连续调用。 transform 方法就是把一个 Stream 作为输入，然后经过计算或数据转换，输出为另一个 Stream。另一个 Stream 中的数据类型可以不同于原类型，数据多少也可以不同（比如实现一个数据的 buffer ）。

transform 的方法签名是：
  Stream transform(StreamTransformer<T, dynamic> streamTransformer)

下面我们构造一个 StreamTransformer ，然后使用 Stream 的 transform() 进行转换：

1.   var transformer = new StreamTransformer<int, int>(
   
2.       handleData: (e, sink){
   
3.         sink.add(e*2);
   
4.       }
   
5.   );
   
6.   stream.transform(transformer).listen(print);
   
7. 
   
8.   // equivalent below
   
9.   stream.map((e) => e*2).listen(print);

复制代码

或

1.   class MyTransformer extends StreamEventTransformer {
   
2.     handleData(e, sink){
   
3.       sink.add(e*2);
   
4.     }
   
5.   }
   
6. 
   
7.   stream.transform(new MyTransformer()).listen(print);

复制代码

使用 StreamTransformer 接口的工厂构造函数 或者 继承 StreamEventTransformer 都可以构造一个 transformer 。其本质和我们处理一个 Stream 是一样的，就要要处理 handleData、handleError 和 handleDone 这三件事。上面的 transform 和 map 方法类似，但是 transform 方法比 map 方法更灵活。map 只能做1对1的转换，而 transform 并没有这个要求，因为它是利用 sink 来添加数据，而不是返回转换结果。transform 方法和 StreamTransformer 接口是一种更通用的设计。

举个更实用点例子，Dart 中的 StringDecoder 和 StringEncoder 就是一个 StreamTransformer，负责实现 byte stream 和 String stream 之间的转换。LineTransformer 是切分行的 transformer。比如，使用 Stream 读文件需要先将字节转换为字符，然后还可以按行读取。

1.   file.openRead()
   
2.       .transform(new StringDecoder())
   
3.       .transform(new LineTransformer())      .listen(your_process);

复制代码

注意，不管是 Stream.map() 还是 Stream.transform() ，他们都是在做转换，而非订阅。对于单模式 Stream ，如果没有添加订阅者，那么转换方法根本不会执行(可能是由于是惰性执行的缘故）。

1. stream.map((e){
   
2.   print(e);
   
3.   return e*2;
   
4. });
   
5. // nothing output, because lazy evaluate
   
6. 
   
7. class MyTransformer extends StreamEventTransformer {
   
8.   handleData(e, sink){
   
9.     print(e);
   
10.     sink.add(e*2);
    
11.   }
    
12. }
    
13. stream.transform(new MyTransformer());
    
14. // nothing output, because no subscription

复制代码

上面的示例中，都在转换过程中做了输出，但实际不会输出内容，因为没有用 listen 添加订阅者。