Dart 异步

2010 年 12 月 31 日

Dart是基于 事件循环机制 的 单线程模型

一条执行线上，同时且只能执行一个任务（事件），其他任务都必须在后面排队等待被执行。也就是说，在一条执行线上，为了不阻碍代码的执行，每遇到的耗时任务都会被挂起放入任务队列，待执行结束后再按放入顺序依次执行队列上的任务，从而达到异步效果。

单线程模型按照代码编写的顺序，自上而下运行，这是我们所认知的，但是当遇到耗时操作（IO/网络请求）等，会给UI造成卡顿阻塞，那么在Flutter中是怎么解决这个问题的呢？接下来我们来仔细分析：

1. ioslate

Dart是基于单线程模型的语言。在Dart中也有自己的进程机制 – isolate 。APP的启动入口main函数就是一个 ioslate ，Dart中的ioslate之间无法直接共享内存，不同ioslate之间只能通过ioslate api进行通信。

在Dart中实现并发可以用Isolate，它是类似于线程(thread)但不共享内存的独立运行的worker，是一个独立的Dart程序执行环境。其实默认环境就是一个main isolate。

在Dart语言中，所有的Dart代码都运行在某个isolate中，代码只能使用所属isolate的类和值。不同的isolate可以通过port发送message进行交流。(首字母大写的Isolate代表Isolate对象，小写的isolate代表一个独立的Dart代码执行环境)

一个Isolate对象就是一个isolate(执行环境)的引用，通常不是当前代码所在的isolate，也就是说，当你使用Isolate对象时，你的目的应该是控制其他isolate，而不是当前的isolate。

import 'dart:isolate';

void main() {
  ReceivePort port = ReceivePort();
  Isolate.spawn(fun, port.sendPort);///固定写法
  port.listen((t) {///这里是设置当前receivePort 监听
    print("接收到其他isolate发过来的消息！");///这里接收了其他isolate发送的消息
    print(t);///接收到的为fun方法里面发送的消息
  });
}
void fun(SendPort sendPort) {
  var receivePort = new ReceivePort();
  var port = receivePort.sendPort;
  port.send("a");///发送消息
  sendPort.send("---");///发送消息
  receivePort.listen((t) {///这里是设置当前receivePort 监听
    print("接收到当前isolate发过来的消息！");///这里接收了当前发送的消息
    print(t);
  });
}

2. Dart消息机制

Dart线程中有一个消息循环机制（event looper）和两个队列（event queue事件队列和microtask queue微服务队列）

event queue 事件队列包含所有外来的事件：IO操作，按钮点击，绘图等消息。任意ioslate中新增的event都会放入消息队列中排队等待
microtask queue 微任务队列值在当前ioslate的任务队列中排队，优先级高于event queue

2.1 Event Looper

Dart代码的运行是从main函数开始的，main函数执行完毕后，Event Looper开始工作，MQ微服务队列优先级高于EQ事件队列，所以Event Looper优先执行MQ中的event事件，当全部执行完毕后，再去执行EQ事件队列中的event。

2.2 Microtask Queue 微服务队列

MQ 微服务队列的优先级要高于EQ事件队列，Event Looper优先执行MQ队列中的事件，其次执行EQ事件队列中的事件
MQ 微服务队列中一般来自于Dart内部，并且微任务非常少。因为如果微任务很多的话，就会造成事件队列排不上对，会阻塞任务队列的执行

创建微服务

可以通过async下的schedlueMicrotask来创建一个微任务：

import "dart:async";

main(List args) {
  scheduleMicrotask(() {
    print("我是一个微任务");
  });
}

2.3 Event Queue 事件队列

事件队列一般来自于外部事件任务，例如IO操作、计时器、点击、绘图等等
上面说过如果微任务很多的话就有可能造成事件队列中的事件排不上对，可能会造成点击一个按钮没有反应造成阻塞，所以微服务不宜过多

另外一部分来源于Future（自定义EQ事件）

2.4 await、async

它们是Dart中的关键字，可以让我们用同步的代码格式来做异步的任务
async 描述一个执行异步操作的方法
await 表示一直等待异步方法返回结果，才继续往后执行
一般一个async的函数会返回一个Future

//HTTP的get请求返回值为Future类型，即其返回值未来是一个String类型的值
  getData() async {    //async关键字声明该函数内部有代码需要延迟执行
    return await http.get(Uri.encodeFull(url), headers: {"Accept": "application/json"}); //await关键字声明运算为延迟执行，然后return运算结果
  }

:warning:注意：这里retrun的并不是我们想要的数据结构类型，他的返回类型时一个await延迟执行的结果。在Dart中，有await标记的运算，其返回结构都是一个Future对象，所以我们可以这样写：

String data;
getData() async {
  data = await http.get(Uri.encodeFull(url), headers: {"Accept": "application/json"});     //延迟执行后赋值给data
}

:warning:：

await关键字必须在async函数内部使用
调用async函数必须使用await关键字

3. Future

Future对象表示异步操作的结果，进程或者IO会延迟完成；我们可以通过它在某个时间点获得异步任务中返回的值，每一个Future都是一个Event，例如我们常用的RefreshIndicator下拉刷新组件中的onRefresh()方法就是一个event，每一个被await标记的句柄也是一个event，没创建一个Future都会把这个Future放进EQ队列中进行排队。

3.1 Future常用函数

then() 函数任务执行完成后会进入then函数，能够获取返回的结果
**catchError()**函数任务失败时，可以在此捕获异常
**whenComplete()**函数任务结束完成后，进入这里
**wait()**函数等待多个异步任务执行完成后，再调用then()
**delayed()**函数延迟任务执行

:warning:：

Future没有执行完成（有任务需要执行），那么then会直接被添加到Future的函数执行体后；
如果Future执行完后就then，该then的函数体被放到如微任务队列，当前Future执行完后执行微任务队列
如果Future世链式调用，意味着then未执行完，下一个then不会执行

// future_1加入到eventqueue中，紧随其后then_1被加入到eventqueue中
Future(() => print("future_1")).then((_) => print("then_1"));

// Future没有函数执行体，then_2被加入到microtaskqueue中
Future(() => null).then((_) => print("then_2"));

// future_3、then_3_a、then_3_b依次加入到eventqueue中
Future(() => print("future_3")).then((_) => print("then_3_a")).then((_) => print("then_3_b"));

3.2 Future使用

Future createFile(String path) async {
  final tempDic = new Directory(path);
  var exits = await tempDic.exists();
  if (exits) {
    return Future(() => false);
  }
  tempDic.createSync(recursive: true);
  return Future(() => true);
}

4. Stream

Stream和 Future 一样都是Dart中用来做异步操作的，官方对其定义为：

Widgets + Stream = Reactive Flutter APP

Stream的作用类似于Android开发中RxJava或者LiveData。它是一个异步流，我们可以在代码中任何地方定义 Stream，然后在其他地方添加数据， Stream 会监听到数据变化，并将改变后的数据传递给监听者。

4.1 Stream分类

单订阅流（Single Subscription）
多订阅流（BroadCast）

4.2 Stream使用

创建一个Stream返回Future：

Stream.fromFuture(xxxx)

创建一个Stream返回集合对象：

Stream.fromIterable(['x','x','x'])

创建一个Stream返回Futures集合对象：

Stream.fromFutures([xxx]);

创建一个Stream返回Duration对象：

Duration interval = Duration(seconds: 1);
Stream stream = Stream.periodic(interval);

详细可见：

https://segmentfault.com/a/1190000019974515

里面有详细的操作符介绍

4.3 StreamController

StreamController类似一个管道，在这个管道中封装了Stream，并向我们提供了两个接口来操作Stream：

sink 从Stream中的一端插入数据
stream 从Stream的另一端弹出数据

具体使用：

创建StreamController

StreamController controller = new StreamController();

向Stream中添加数据

controller.sink.add("Item1");
controller.sink.add("Item2");
controller.sink.add("Item3");

创建Stream监听器

通过StreamController中的stream.listen()，设置监听Stream弹出的数据：

controller.stream.listen((item) => print(item));

// 向Stream中添加error
controller.sink.addError('there is a problem!');
controller.sink.close(); // 调用close方法，结束Stream中的逻辑处理

以上部分是单订阅流，也就是单监听器的Stream，下面来看下多订阅流的使用：

构建多订阅流的方式有两种

直接创建多订阅Stream

StreamController streamController = StreamController.broadcast();
    streamController.stream.listen((data){
      print(data);
    },onError: (error){
      print(error.toString());
    });
    streamController.stream.listen((data) => print(data));
    streamController.add("bbb");

将单订阅流转成多订阅流

StreamController streamController = StreamController();
  Stream stream =streamController.stream.asBroadcastStream();
  stream.listen((data) => print(data));
  stream.listen((data) => print(data));
  streamController.sink.add("aaa");
  streamController.close();

4.4 StreamBuilder使用

StreamBuilder是Flutter中的一个Widget，记录着流中最新的数据，当数据流发生变化时，会自动调用Builder进行重建

const StreamBuilder({
    Key key,
    this.initialData,
    Stream stream,
    @required this.builder,
  }) : assert(builder != null),
       super(key: key, stream: stream);

可以看到StreamBuilder需要接受一个Stream

使用 StreamController 结合 StreamBuider 对官方的计数器进行改进，取代setState刷新页面，代码如下

class MyHomePage extends StatefulWidget {
  @override
  _MyHomePageState createState() => _MyHomePageState();
}

class _MyHomePageState extends State {
  int _count = 0;
  final StreamController _streamController = StreamController();

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      body: Container(
        child: Center(
          child: StreamBuilder(
              stream: _streamController.stream,
              builder: (BuildContext context, AsyncSnapshot snapshot) {
                return snapshot.data == null
                    ? Text("0")
                    : Text("${snapshot.data}");
              }),
        ),
      ),
      floatingActionButton: FloatingActionButton(
          child: const Icon(Icons.add),
          onPressed: () {
            _streamController.sink.add(++_count);
          }),
    );
  }

  @override
  void dispose() {
    _streamController.close();
    super.dispose();
  }
}

参考：

https://www.cnblogs.com/lxlx1798/p/11126564.html

M	T	W	T	F	S	S
« Jan
1	2	3	4	5	6	7
8	9	10	11	12	13	14
15	16	17	18	19	20	21
22	23	24	25	26	27	28
29	30

演道网

Dart 异步

1. ioslate

2. Dart消息机制

2.1 Event Looper

2.2 Microtask Queue 微服务队列

创建微服务

2.3 Event Queue 事件队列

2.4 await、async

3. Future

3.1 Future常用函数

3.2 Future使用

4. Stream

4.1 Stream分类

4.2 Stream使用

4.3 StreamController

4.4 StreamBuilder使用

About The Author

shine

1. ioslate

2. Dart消息机制

2.1 Event Looper

2.2 Microtask Queue 微服务队列

创建微服务

2.3 Event Queue 事件队列

2.4 await、async

3. Future

3.1 Future常用函数

3.2 Future使用

4. Stream

4.1 Stream分类

4.2 Stream使用

4.3 StreamController

4.4 StreamBuilder使用

Related Posts

About The Author

shine