Dart 教程 - Dart 异步支持 - 《Dart》

Dart的Event Loop
Dart的Event Queue
单线程模型
Event Queue以及Microtask Queue
- Microtask Queue存在的意义
- Microtask Queue && Event Queue执行流程图
异步任务调度
Future
async/await
Stream
Stream的种类
- “Single-subscription” streams 单订阅流
- “Broadcast” streams 多订阅流

Dart是单线程模型，也就没有了所谓的主线程/子线程之分 Dart也是 Event-Looper 事件循环以及 Event-Queue 事件梯队的模型所有的事件都是通过 EventLooper事件循环程序 依次执行

Dart的Event Loop

从EventQueue中获取Event
处理Event
直到EventQueue为空

Event Queue

Dart的Event Queue

而这些Event包括了用户输入，点击，Timer，文件IO等

Event Type

单线程模型

一旦某个Dart的函数开始执行，它将执行到这个函数结束，也就是Dart的函数不会被其他Dart代码打断
Dart中没有线程的概念，只有isolate，每个isolate都是隔离的，并不会共享内存
而一个Dart程序是在Main isolate 的main函数开始，而在Main函数结束后，Main isolate线程开始一个一个（one by one）的开始处理Event Queue中的每一个Event

Main Isolate

Event Queue以及Microtask Queue

Dart中的 Main Isolate 只有一个Event Looper
但是存在两个Event Queue
- Event Queue
- Microtask Queue

Microtask Queue存在的意义

希望通过这个Queue来处理稍晚一些的事情，但是在下一个消息到来之前需要处理完的事情

Microtask Queue && Event Queue执行流程图

当Event Looper正在处理Microtask Queue中的Event时候，Event Queue中的Event就停止了处理了，此时App不能绘制任何图形，不能处理任何鼠标点击，不能处理文件IO等等
Event-Looper挑选Task的执行顺序为：

优先全部执行完Microtask Queue中的Event
直到Microtask Queue为空时，才会执行Event Queue中的Event

Dart中只能知道Event处理的先后顺序，但是并不知道某个Event执行的具体时间点，因为它的处理模型是一个单线程循环，而不是基于时钟调度（即它的执行只是按照Event处理完，就开始循环下一个Event，而与Java中的Thread调度不一样，没有时间调度的概念），也就是我们既是指定另一个Delay Time的Task，希望它在预期的时间后开始执行，它有可能不会在那个时间执行，需要看是否前面的Event是否已经Dequeue

异步任务调度

当有代码可以在后续任务执行的时候，有两种方式，通过dart:async这个Lib中的API即可：

使用 Future 类，可以将任务加入到Event Queue的队尾
使用 scheduleMicrotask函数(安排微任务)，将任务加入到Microtask Queue队尾

当使用EventQueue时，需要考虑清楚，尽量避免microtask queue过于庞大，否则会阻塞其他事件的处理

Use Event Queue

Future

Future是一个实现异步的类，跟JS中Promise有点类似

Future构造函数

一般常用的Future构造函数：

Future<T> Future(FutureOr<T> Function() computation)

main(List<String> args) {
  var futureInstance = Future<String>(() => "12345");
  futureInstance.then((res) {
    print(res);
  }).catchError((err) {
    print(err);
  });
}

Future.value

main(List<String> args) {
var futureInstance = Future.value(1234);
futureInstance.then((val)=>print(val));
}
//输出结果 1234

Future.then

而一般常用的还有当有分治任务时，需要将一个大任务拆成很多小任务一步步执行时，就需要使用到Future.then函数来拆解任务

void main() {
  new Future(() => futureTask) //  异步任务的函数
      .then((m) => {print("futueTask execute result:$m")}) //任务执行完后的子任务
      .then((m){print(m);return 10000000;})  //其中m为上个任务执行完后的返回的结果
      .then((m) => printLength(m))
      .whenComplete(() => whenTaskCompelete); //当所有任务完成后的回调函数

  futureTask();
}

int futureTask() {
  return 21;
}

void printLength(int number) {
  print("Text Length:${number.toString().length}");
}

void whenTaskCompelete() {
  print("Task Complete");
}

[Running] dart "/Users/kim/test/main.dart"
futueTask execute result:Closure: () => int from Function 'futureTask': static.
{null}
Text Length:8
[Done] exited with code=0 in 0.209 seconds

Future.error

main(List<String> args) {
  var futureInstance = Future.error("这是一个错误");
  futureInstance.catchError((err) {
    print(err);
  });
}

//输出结果 这是一个错误

Future.catchError

如果异步任务发生错误，我们可以在 catchError 中捕获错误，我们将上面示例改为：

Future.delayed(new Duration(seconds: 2),(){
   //return "hi world!";
   throw AssertionError("Error");  
}).then((data){
   //执行成功会走到这里  
   print("success");
}).catchError((e){
   //执行失败会走到这里  
   print(e);
});

在本示例中，我们在异步任务中抛出了一个异常，then的回调函数将不会被执行，取而代之的是 catchError回调函数将被调用；但是，并不是只有 catchError回调才能捕获错误，then方法还有一个可选参数 onError，我们也可以它来捕获异常：

Future.delayed(new Duration(seconds: 2), () {
    //return "hi world!";
    throw AssertionError("Error");
}).then((data) {
    print("success");
}, onError: (e) {
    print(e);
});

Future.delayed

当任务需要延迟执行时，可以使用new Future.delay来将任务延迟执行，而如上所述，只有当Main isolate的Event Queue处于Idle 闲置的状态时，才会延迟1s执行，否则等待的时间会比1s长很多

Future.delayed(const Duration(seconds: 1), () => futureTask);

import 'dart:async';

main(List<String> args) {
  //———————————————— 方法1 ————————————————
  Future delay(int ms) {
    var com = Completer();
    Timer(Duration(milliseconds: ms), () {
      com.complete("方法1");
    });
    return com.future;
  }

  delay(1000).then((res) {
    print(res);
  });

  //———————————————— 方法2 ————————————————
  Future.delayed(Duration(milliseconds: 2000), () {
    print("方法2");
  });
}

Future.whenComplete

有些时候，我们会遇到无论异步任务执行成功或失败都需要做一些事的场景，比如在网络请求前弹出加载对话框，在请求结束后关闭对话框。这种场景，有两种方法:

第一种是分别在 then 或 catch 中关闭一下对话框

第二种就是使用Future的 whenComplete 回调

Future.delayed(new Duration(seconds: 2),(){
 //return "hi world!";
 throw AssertionError("Error");
}).then((data){
 //执行成功会走到这里 
 print(data);
}).catchError((e){
 //执行失败会走到这里   
 print(e);
}).whenComplete((){
 //无论成功或失败都会走到这里
});

Future.wait

有些时候，我们需要等待多个异步任务都执行结束后才进行一些操作，比如我们有一个界面，需要先分别
从两个网络接口获取数据，
获取成功后，我们需要将两个接口数据进行特定的处理后再显示到UI界面上
应该怎么做？答案是 Future.wait，它接受一个Future数组参数，只有数组中所有Future都执行成功后，才会触发then的成功回调，只要有一个Future执行失败，就会触发错误回调。
下面，我们通过模拟 Future.delayed 来模拟两个数据获取的异步任务，等两个异步任务都执行成功时，将两个异步任务的结果拼接打印出来，代码如下：

Future.wait([
  // 2秒后返回结果  
  Future.delayed(new Duration(seconds: 2), () {
    return "hello";
  }),
  // 4秒后返回结果  
  Future.delayed(new Duration(seconds: 4), () {
    return " world";
  })
]).then((results){
  print(results[0]+results[1]);
}).catchError((e){
  print(e);
});


/* 
main(List<String> args) {
  Future future1() async { print("这是异步函数1"); }
  Future future2() async { print("这是异步函数2"); }

  var futureList = <Future>[];
  futureList.add(future1());
  futureList.add(future2());
  Future.wait(futureList);
} 
*/

执行上面代码，4秒后你会在控制台中看到 “hello world”
与JavaScript中的 Promise.all 相似

Future.sync

同步运行的任务: 指的是构造Future的时候传入的函数是同步运行的,当前传递的函数可以一起执行，和then不同的是，then回调进来的函数是调度到微任务队列异步执行的

Future.sync((){

});

main(List<String> args) {
  Future.sync(syncFn).then((m) {
    print(m);
  });
}

int syncFn() {
  return 123456789;
}

//123456789

Future.microtask

future内部就是调用了 scheduleMicrotask 函数，用来将当前任务加入到microtask-queue中，实现’插队’功能

main(List<String> args) {
  Future.microtask(() {
    //优先执行的业务逻辑
  });
}

使用scheduleMicrotask

在最顶层的调用关系中，使用该函数即可

async.scheduleMicrotask(() => microtask());

void microtask(){
  //  doing something
}

总结

当需要做动画的时候，不要使用Future，而需要使用 animateFrame

Future<num> get animationFrame {
  var completer = new Completer<num>.sync();
  requestAnimationFrame((time) {
    completer.complete(time);
  });
  return completer.future;
}

Future中的then并没有创建新的Event丢到Event Queue中，而只是一个普通的Function Call，在FutureTask执行完后，立即开始执行
当Future在then函数先已经执行完成了，则会创建一个task，将该task的添加到中，并且该任务将会执行通过then传入的函数
Future只是创建了一个Event，将Event插入到了Event Queue的队尾
使用Future.value构造函数的时候，就会和第二条一样，创建Task丢到 microtask Queue 中执行then传入的函数
Future.sync构造函数执行了它传入的函数之后，也会立即创建Task丢到 microtask Queue 中执行

async/await

Dart 库中包含许多返回 Future 或 Stream 对象的函数. 这些函数在设置完耗时任务（例如 I/O 曹组）后，就立即返回了，不会等待耗任务完成。使用 async 和 await 关键字实现异步编程。可以让你像编写同步代码一样实现异步操作。

处理 Future

可以通过下面两种方式，获得 Future 执行完成的结果：

使用 async 和 await
使用 Future API，具体描述，参考库概览

使用 async 和 await 关键字的代码是异步的。虽然看起来有点想同步代码。例如，下面的代码使用 await 等待异步函数的执行结果。

await lookUpVersion();

多任务处理

同JavaScript的Promise.all()的用法

if (await Permission.contacts.request().isGranted) {
  // Either the permission was already granted before or the user just granted it.
}

// You can request multiple permissions at once.
Map<Permission, PermissionStatus> statuses = await [
  Permission.location,
  Permission.storage,
].request();
print(statuses[Permission.location]);

见 https://pub.flutter-io.cn/packages/permission_handler#-installing-tab-

要使用 await ，代码必须在异步函数（使用 async 标记的函数）中：

Future checkVersion() async {
  var version = await lookUpVersion();
  // Do something with version
}

⚠️注意: 虽然异步函数可能会执行耗时的操作，但它不会等待这些操作。相反，异步函数只有在遇到第一个 await 表达式（详情见）时才会执行。也就是说，它返回一个 Future 对象，仅在await表达式完成后才恢复执行。

使用 try， catch，和 finally 来处理代码中使用 await 导致的错误。

try {
  version = await lookUpVersion();
} catch (e) {
  // React to inability to look up the version
}

在一个异步函数中可以多次使用 await 。例如，下面代码中等待了三次函数结果：

var entrypoint = await findEntrypoint();
var exitCode = await runExecutable(entrypoint, args);
await flushThenExit(exitCode);

在 await表达式中，表达式的值通常是一个 Future 对象；如果不是，这是表达式的值会被自动包装成一个 Future 对象

Future 对象指明返回一个对象的承诺（类似promise）
await 表达式执行的结果为这个返回的对象。
await 表达式会阻塞代码的执行，直到需要的对象返回为止。

如果在使用 await 导致编译时错误，确认 await 是否在一个异步函数中。例如，在应用的 main()函数中使用 await ， main() 函数的函数体必须被标记为 async

Future main() async {
  checkVersion();
  print('In main: version is ${await lookUpVersion()}');
}

声明异步函数

函数体被 async 标示符标记的函数，即是一个异步函数。将 async 关键字添加到函数使其返回Future。例如，考虑下面的同步函数，它返回一个 String ：

String lookUpVersion() => '1.0.0';

例如，将来的实现将非常耗时，将其更改为异步函数，返回值是 Future 。

Future<String> lookUpVersion() async => '1.0.0';

⚠️注意: 函数体不需要使用Future API。如有必要， Dart 会创建 Future 对象。如果函数没有返回有效值，需要设置其返回类型为Future<void>

处理 Stream

当需要从 Stream 中获取数据值时，可以通过一下两种方式：

使用 async 和一个异步循环(await for)
使用 Stream API, 更多详情，参考 in the library tour

⚠️提示: 在使用 await for 前，确保代码清晰，并且确实希望等待所有流的结果。例如，通常不应该使用 await for 的UI事件侦听器，因为UI框架会发送无穷无尽的事件流。

以下是异步for循环的使用形式：

await for (varOrType identifier in expression) {
  // Executes each time the stream emits a value.
  // 每次流发出值时执行
}

上面表达式返回的值必须是 Stream 类型。执行流程如下：

等待，直到流发出一个值。
执行 for 循环体，将变量设置为该发出的值
重复1和2，直到关闭流。

使用 break return 语句可以停止接收 stream 的数据，这样就跳出了 for 循环，并且从 stream 上取消注册。

如果在实现异步 for 循环时遇到编译时错误，请检查确保 await for 处于异步函数中。
例如，要在应用程序的 main() 函数中使用异步 for循环， main() 函数体必须标记为 async
```
Future main() async {
// ...
await for (var request in requestServer) {
  handleRequest(request);
}
// ...
}
```
有关异步编程的更多信息，请参考 dart:async 部分。同时也可参考文章 Dart Language Asynchrony Support: Phase 1 和 Dart Language Asynchrony Support: Phase 2, 以及 Dart language specification

Stream

Stream非常有特点但却不太好理解，我与其按照字面意思把它看作流，更愿意把它看成一个工厂或者是机器

我们来看看这个机器它有什么特点：

它有一个入口，可以放东西/指令(anything),这个机器不知道入口什么时候会放东西进来
中间的机器能够生产或者加工，这应该会耗费一些时间
它有一个出口，应该会有产品从那出来,我们也不知道到底什么时候产品会从出口出来

整个过程，时间都是一个不确定因素，我们随时都可以向这个机器的入口放东西进去，放进去了以后机器进行处理，但是我们并不知道它多久处理完。所以出口是需要专门派人盯着的，等待机器流出东西来。整个过程都是以异步的眼光来看的

我们将机器模型转化成Stream

这个大机器就是StreamController，它是创建流的方式之一
StreamController有一个入口，叫做sink, 它可以使用add方法放东西进来，放进去以后就不再关心了
StreamController有一个出口，叫做stream, 机器处理完毕后就会把产品从出口丢出来，但是我们并不知道什么时候会出来，所以我们需要使用 listen 方法一直监听这个出口。而且当多个物品被放进来了之后，它不会打乱顺序，而是先入先出

Stream操作也是dart中提供的用来处理异步操作的工具，和Future不同的是它可以接收多个异步操作的结果(无论成功或失败)
我们可以理解为:执行异步任务时，可以通过多次触发成功或失败事件来传递结果数据或错误异常
例如:我们开发的过程中经常见到的场景:多个文件读写，网络下载可能会发起多个等，接下来我们根据一个案例分析一下Stream常用的操作:

void main(){
   Stream.fromFutures([loginFuture,getUserInfoFuture,saveUserInfoFuture])
         .listen((data){
          //各个任务返回的结果的回调
         },onError:((err){
           //各个任务执行失败的回调
         }),
         onDone : ((){
           //监听各个任务执行的过程中完成的时候的回调
         })
         )
         .onDone((){
           //所有任务全部完成的回调
         });
}

Future loginFuture = Future<String>((){
   //这里调用登录操作
   login('admin','123456');
});
String login(String userName, String pwd){
  //登录操作
}
bool getUserInfo(int id){
  //获取用户信息
}
Future<String> getUserInfoFuture =Future((){
  getUserInfo(1);
});

Future saveUserInfoFuture = Future((){
   saveUserInfo("admin");
});

void saveUserInfo(String userInfo){

}

可以看到 listen 方法中我们可以对每一个任务进行精细的回调处理，甚至所有任务执行完毕以后，我们还有
cancel
pause
resume
onError
onDone
等回调可以分别对整个组任务执行过程的不同阶段进行精细度的处理
在上面我们使用了fromFutures方法使用Stream,除此之外:

Stream使用过程中我们也经常用fromFuture 方法用来处理单个Futrue/fromIterable 方法处理集合中的数据
当然，除了这种常规的Stream操作以外，dart还提供了两个专门操作/创建流的类，可以实现流操作的复杂操作

获得 Stream 的方法

通过构造函数
使用StreamController
IO Stream

Stream有三个构造方法

Stream.fromFuture 从Future创建新的单订阅流,当future完成时将触发一个data或者error，然后使用Down事件关闭这个流
Stream.fromFutures 从一组Future创建一个单订阅流，每个future都有自己的data或者error事件，当整个Futures完成后，流将会关闭。如果Futures为空，流将会立刻关闭。
Stream.fromIterable 创建从一个集合中获取其数据的单订阅流
```
Stream.fromIntreable([1,2,3]);
```

StreamController

如果你想创建一条新的流的话，非常简单！
使用StreamController，它为你提供了非常丰富的功能😀 ，你能够在streamController上发送数据，处理错误，并获得结果！

//任意类型的流
StreamController controller = StreamController();
controller.sink.add(123);
controller.sink.add("xyz");
controller.sink.add(()=>print("输出"));

//创建一条处理int类型的流
StreamController<int> numController = StreamController();
numController.sink.add(123);

泛型定义了我们能向流上推送什么类型的数据。它可以是任何类型！

你需要将一个方法交给stream的 listen函数，这个方法入参(data)是我们的StreamController处理完毕后产生的结果，我们监听出口，并获得了这个结果(data)。这里可以使用lambda表达式，也可以是其他任何函数。

import 'dart:async';

void main(List<String> args) {
  StreamController controller = StreamController();
  controller.sink.add(123);
  controller.sink.add(() => print("添加一个函数"));

  //监听这个流的出口，当有data流出时，打印这个data
  StreamSubscription subscription = controller.stream.listen((data) {
    if (data is Function) {
      data();
    } else {
      print("$data");
    }
  });
  controller.sink.add(() => print("最后执行的"));
}

/* 
123
添加一个函数
最后执行的
*/

通过 async* 生成 stream

如果我们有一系列事件需要处理，我们也许会需要把它转化为 stream。这时候可以使用 async - yield* 来生成一个 Stream

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

当循环退出时，这个 Stream 也就 done 了。我们可以结合之前说的 await for 更加深刻的体验一下

import 'dart:async';

void main() async {
  var stream = countStream(100);
  var sum = await sumStream(stream);
  print(sum);
}

Future<int> generateData(int data) async => data;

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

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

/*
5050
*/

点击这里直接运行上面样例代码
点击这里查看JavaScript中的Promise和yeild

监听 Stream 的方法

监听一个流最常见的方法就是listen。当有事件发出时，流将会通知listener
Listen方法提供了这几种触发事件：

onData(必填) 收到数据时触发
onError 收到Error时触发
onDone 结束时触发
unsubscribeOnError 遇到第一个Error时是否取消订阅，默认为false

使用 await for 处理 Stream

除了通过 listen，我们还可以使用 await for 来处理

import 'dart:async';

void main() async {
  var stream = countStream(100);
  var sum = await sumStream(stream);
  print(sum);
}

Future<int> generateData(int data) async => data;

Stream<int> countStream(int to) async* {
  for (var i = 0; i < 100000; i++) { print(i); }
  for (int i = 1; i <= to; i++) {
    yield await generateData(i);
  }
}

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

/*
...
打印完99999才输出
5050
*/

这段代码将会接收一个 Stream 然后统计所有事件之和，然后返回结果
await for 能够在每个事件到来的时候处理它
我们知道，一个Stream 它接收事件的时机是不确定的，那什么时候应该退出await for循环呢？
答案是，当这个Stream完成或关闭的时候

转换清洗现有流

假如你已经有了一个流(Transforming an existing stream)，你可以通过它转化成为一条新的流。非常简单！流提供了: map() where() expand() take() 方法，能够轻松将已有的流转化为新的流

where

如果你想要筛选掉一些不想要的事件。例如一个猜数游戏，用户可以输入数字，当输入正确的时候，我们做出一定反应。而我们必须筛选掉所有错误的答案，这个时候我们可以使用where筛选掉不需要的数字

(stream对象).where((event){...})

where函数接收一个事件，每当这个流有东西流到where函数的时候，这就是那个事件。我们或许根本不需要这个事件，但是必须作为参数传入,之后可以筛选符合要求的数据返回

take

如果你想要控制这个流最多能传多少个东西。比如输入密码，我们可能想让用户最多输四次，那么我们可以使用take来限制

.take(4);

take函数接收一个int，代表最多能经过take函数的事件次数。当传输次数达到这个数字时，这个流将会关闭，无法再传输

transform

如果你需要更多的控制转换，那么请使用transform()方法。他需要配合StreamTransformer进行使用。我们先来看下面一段猜数游戏，然后我会向你解释

import 'dart:async';

main(List<String> args) {

  StreamController<int> controller = StreamController<int>();
  final transformer =
      StreamTransformer<int, String>.fromHandlers(handleData: (value, sink) {
    if (value == 100) {
      sink.add("你猜对了");
    } else {
      sink.addError('还没猜中，再试一次吧');
    }
  });

  controller.stream
      .transform(transformer)
      .listen((data) => print(data), onError: (err) => print(err));

  controller.sink.add(23);
  controller.sink.add(100);
}

/*
还没猜中，再试一次吧
你猜对了
*/

StreamTransformer<S,T> 是我们stream的检查员，他负责接收stream通过的信息，然后进行处理返回一条新的流。

S 代表之前的流的输入类型，我们这里是输入一个数字，所以是int。
T 代表转化后流的输入类型，我们这里add进去的是一串字符串，所以是String。
handleData 接收一个value并创建一条新的流并暴露sink，我们可以在这里对流进行转化。
我们还可以addError进去告诉后面有问题

然后我们监听transform之后的流，当转换好的event流出时，我们打印这个event，这个event就是我们刚才add进sink的数据。onError能够捕捉到我们add进去的err。

Stream的种类

“Single-subscription” streams 单订阅流

单个订阅流在流的整个生命周期内仅允许有一个listener。它在有收听者之前不会生成事件，并且在取消收听时它会停止发送事件，即使你仍然在Sink.add更多事件。
即使在第一个订阅被取消后，也不允许在单个订阅流上进行两次侦听
单订阅流通常用于流式传输更大的连续数据块，如文件I/O

import 'dart:async';

main(List<String> args) {
  StreamController controller = StreamController();
  controller.stream.listen((data) => print(data));
  controller.stream.listen((data) => print(data));
  controller.sink.add(123);
}


/*
Ken-MacBook:JavaScript kim$ dart "/Users/Ken/JavaScript/tempCodeRunnerFile.dart"
Unhandled exception:
Bad state: Stream has already been listened to.
#0      _StreamController._subscribe (dart:async/stream_controller.dart:670:7)
#1      _ControllerStream._createSubscription (dart:async/stream_controller.dart:820:19)
#2      _StreamImpl.listen (dart:async/stream_impl.dart:474:9)
#3      main (file:///Users/Ken/JavaScript/tempCodeRunnerFile.dart:6:21)
#4      _startIsolate.<anonymous closure> (dart:isolate-patch/isolate_patch.dart:305:32)
#5      _RawReceivePortImpl._handleMessage (dart:isolate-patch/isolate_patch.dart:174:12)
Ken-MacBook:JavaScript kim$ 
*/

“Broadcast” streams 多订阅流

广播流允许任意数量的收听者，且无论是否有收听者都能产生事件。所以中途进来的收听者将不会收到之前的消息
如果多个收听者想要收听单个订阅流，请使用asBroadcastStream在非广播流之上创建广播流
如果在触发事件时将收听者添加到广播流，则该侦听器将不会接收当前正在触发的事件。如果取消收听，收听者会立即停止接收事件
一般的流都是单订阅流。从Stream继承的广播流必须重写isBroadcast 才能返回true

import 'dart:async';

main(List<String> args) {
  StreamController controller = StreamController();
    //将单订阅流转化为广播流
  Stream stream = controller.stream.asBroadcastStream();
  stream.listen((data) => print(data));
  stream.listen((data) => print(data));
  controller.sink.add(123);
}

//输出：123 123