Dart 异步

Posted 2022-03-10 Hankkin_Coding

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<String> args) 
  scheduleMicrotask(() 
    print("我是一个微任务");
  );

2.3 Event Queue 事件队列

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

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

2.4 await、async

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

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

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

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

⚠️：

• 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()**函数 延迟任务执行

⚠️：

• 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<bool> 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<String>.fromFuture(xxxx)

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

Stream<String>.fromIterable(['x','x','x'])

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

Stream<String>.fromFutures([xxx]);

创建一个Stream返回Duration对象：

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

详细可见：

https://segmentfault.com/a/1190000019974515

里面有详细的操作符介绍

4.3 StreamController

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

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

具体使用：

创建StreamController

StreamController<String> controller = new StreamController<String>();

向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<String> streamController = StreamController.broadcast();
      streamController.stream.listen((data)
        print(data);
      ,onError: (error)
        print(error.toString());
      );
      streamController.stream.listen((data) => print(data));
      streamController.add("bbb");
  
  

• 将单订阅流转成多订阅流

   StreamController<String> 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<T> 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<MyHomePage> 
  int _count = 0;
  final StreamController<int> _streamController = StreamController();

  @override
  Widget build(BuildContext context) 
    return Scaffold(
      body: Container(
        child: Center(
          child: StreamBuilder<int>(
              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