Dart学习4-异步编程

在dart中不存在线程的概念，也就没有锁类似的概念。如果有耗时操作，就需要引入Isolate机制

1 Isolate机制

void main() {
  // Isolate有一个静态方法。传入的第一个参数是一个方法，这边使用的匿名方法。第二个参数就是传入方法的参数
  Isolate.spawn((message) {
    // 这里执行的代码都是子isolate中
    sleep(const Duration(seconds: 10));
  }, "我说从isolate传递到子isolate的数据");
}

那Isolate之间是怎么交互的呢？主isolate和子isolate的资源不共享

void main() {
  // 创建一个消息接收器
  var mainReceivePort = ReceivePort();
  // 从消息接收器中读取消息
  mainReceivePort.listen((message) {
    // 这里接收从子isolate发来的消息
    print("主isolate接收器接收到消息：$message");

    // 如果消息类型是SendPort，则说明是发送器。
    if (message is SendPort) {
      message.send("我是从主isolate发送到子isolate的消息");
    }
  });

  mainReceivePort.sendPort.send("我是从主isolate发来的消息");

  // 创建一个子isolate
  Isolate.spawn((SendPort mainSendPort) {
    // 用主isolate发来的发送器发送消息就能够在主isolate中接收到消息
    mainSendPort.send("我是从子isolate发送到主isolate的消息");

    // 如果我们需要从主isolate向子isolate中发送消息，那就需要在子isolate中创建一个自己的接收器
    var childReceivePort = ReceivePort();
    // 同样的，给子isolate中的接收器设置监听
    childReceivePort.listen((message) {
      print("子isolate接收器接收到消息：$message");
    });
    // 将子isolate中的发送器发送给主isolate，那样就完成了双向连接
    var childSendPort = childReceivePort.sendPort;
    mainSendPort.send(childSendPort);
  }, mainReceivePort.sendPort); // 将消息接收器中的发送器发送给子isolate
}

ReceivePort不用时需要关闭，否则程序不会结束。

receivePort.close();

2 任务队列

在dart中是由任务驱动的。
同android中handler类似，在dart运行环境中也是靠事件驱动的，通过event loop不停的从队列中获取消息或者事件来驱动整个应用的运行，isolate发过来的消息就是通过loop处理。
但是不同的是在android中每个线程只有一个looper所对应的messageQueue，
而dart中有两个队列，一个叫做event queue(事件队列)，另一个叫做microtask queue（微任务队列）
looper会先从微任务队列中获取任务，直到获取完，才去执行事件队列中的任务。
每当执行完事件队列中的一个任务时，又会去检查微任务队列中是否有了任务，如果有，先去执行微任务。

微任务的执行优先级比事件队列任务高

void main() {
  // 在dart中是由任务驱动的。
  // 同android中handler类似，在dart运行环境中也是靠事件驱动的，通过event loop不停的从队列中获取消息或者事件来驱动整个应用的运行，isolate发过来的消息就是通过loop处理。
  // 但是不同的是在android中每个线程只有一个looper所对应的messageQueue，
  // 而dart中有两个队列，一个叫做event queue(事件队列)，另一个叫做microtask queue（微任务队列）
  // looper会先从微任务队列中获取任务，直到获取完，才去执行事件队列中的任务。
  // 每当执行完事件队列中的一个任务时，又会去检查微任务队列中是否有了任务，如果有，先去执行微任务。微任务的执行优先级比事件队列任务高
  print("开始执行main方法--- ${DateTime.now()}");

  var receivePort = ReceivePort();
  receivePort.listen((message) {
    print("执行$message ${DateTime.now()}");
  });

  receivePort.sendPort.send("事件任务1");
  Future.microtask(() {
    // 执行微任务
    print("执行微任务1 ${DateTime.now()}");

    // 这边休眠2s，会阻塞。又佐证了在dart中是单线程的。
    sleep(const Duration(seconds: 2));
  });
  receivePort.sendPort.send("事件任务2");
  Future.microtask(() {
    // 执行微任务
    print("执行微任务2 ${DateTime.now()}");
  });
  receivePort.sendPort.send("事件任务3");

  // 如果在这里休眠3s，那上面的任务队列中的任务也都要在3s后才执行
  // 只有当main任务执行完成之后，才会执行任务队列中的任务
  sleep(const Duration(seconds: 3));

  print("main方法执行结束。。。 ${DateTime.now()}");
}

上面的执行打印结果：

开始执行main方法--- 2022-03-23 16:35:37.269329
main方法执行结束。。。 2022-03-23 16:35:40.287288 ----------------3s后才开始执行任务队列中的任务
执行微任务1 2022-03-23 16:35:40.288894 -------------------------------执行微任务1，2s后才开始继续执行任务
执行微任务2 2022-03-23 16:35:42.291514
执行事件任务1 2022-03-23 16:35:42.294005
执行事件任务2 2022-03-23 16:35:42.294192
执行事件任务3 2022-03-23 16:35:42.294231

3 Future

future意为未来，就是将来要执行的事情。需要main方法执行结束后才能执行，所以不会造成阻塞。

void main() {
  // 延迟处理的消息。至少3s后才能执行。
  Future.delayed(Duration(seconds: 1), () {
    // 读取路径下文件的内容，返回的结果类型Future<String>
    Future<String> readAsString =
        File(r"/Users/phoenix/Downloads/test1").readAsString();

    // 通过then来获取到泛型的值。then也有返回值FutureOr<R>，FutureOr是可以返回void，也可以返回Future<R>
    // 这个是模拟返回一个int值。int值可以继续通过then来获取。
    Future<int> then = readAsString.then((value) {
      print("value=$value");
      return 100;
    });

    // 其实我们可以写成链式调用
    File(r"/Users/phoenix/Downloads/test1").readAsString().then((value) {
      print("链式调用 value=$value");
    });
  });
}

4 Stream

Stream（流）在dart中也经常出现，表示发出的一些列的异步数据。

Future表示稍后获得的一个数据，所有的异步的操作的返回值都用future来表示。但是Future只能表示一次异步获得的数据。而Stream表示多次异步获得的数据。

比如IO处理的时候，每次只会读取一部分的数据和一次性读取整个文件相比，Stream的好处是处理过程中内存占用较小。而File.readString()，是一次性读取整个文件的内容，文件很大的话会导致内存占用过大的问题

import 'dart:io';

void main() {
  // 通过流的方式来读写
  File file = File(r"/Users/phoenix/Downloads/text.epub");
  
  // openRead([int? start, int? end]) 可以传入读取的坐标，也可以不传
  // 返回的结果就是对应流的集合
  Stream<List<int>> openRead = file.openRead();
  
  // 设置读的监听
  var listen = openRead.listen((event) {
    // 一次读取65536。读取大文件时，有可能会读取多次
    print("listen(),会调用多次 ${event.length}");
  });
  
  // onData 可以替换掉listen方法
  listen.onData((data) {
    print("onData(),替换掉listen(),会调用多次");
  });

  // 文件读完毕之后的回调
  listen.onDone(() {
    print("文件都读完了");
  });

  // 也可以控制流暂停和重启
  // listen.pause();
  // listen.resume();
}

通过stream进行读写操作

void main() {
  // 通过流的方式来读写
  File file = File(r"/Users/phoenix/Downloads/text.epub");
  File outFile = File(r"/Users/phoenix/Downloads/text2.epub");

  // openRead([int? start, int? end]) 可以传入读取的坐标，也可以不传
  // 返回的结果就是对应流的集合
  Stream<List<int>> openRead = file.openRead();
  var openWrite = outFile.openWrite();
  
  // 方式一：我们可以直接全部写入
  openWrite.addStream(openRead);
  
  // 方式二：通过设置读的监听，分段多次写入
  var listen = openRead.listen((event) {
    openWrite.add(data);
  });
}

stream不能直接设置多个监听，那怎么设置多个监听呢？

方式1：通过BroadcastStream

void main() {
  var openRead = File(r"/Users/phoenix/Downloads/test1").openRead();

  // 调用多次会报以下异常
  // FileSystemException: An async operation is currently pending, path = '/Users/phoenix/Downloads/test1'
  // openRead.listen((event) {});
  // openRead.listen((event) {});

  // 那怎么设置多个监听呢？通过广播模式
  var asBroadcastStream = openRead.asBroadcastStream();
  asBroadcastStream.listen((event) {
    print("listen1 event=$event");
  });
  asBroadcastStream.listen((event) {
    print("listen2 event=$event");
  });
}

方式1：通过StreamController

void main() {
  var streamController = StreamController.broadcast();
  
  // 在发送消息之前设置监听，可以接收到广播
  streamController.stream.listen((event) {
    print("1....接收 $event");
  });
  
  // 发送消息
  streamController.add("消息1");
  
  // 在发送消息之后才去设置监听，不可以接收到广播
  streamController.stream.listen((event) {
    print("2....接收 $event");
  });
}

5 async-await

使用async和await的代码都是异步的，但是看起来很像同步代码。当我们需要获取A的结果，再执行B时，你需要then()->then()，但是利用async与await能够很好的解决回调问题。

• async

  被async修饰的方法就变成了异步方法。返回值是Future或者void。可以通过then获取到方法的返回值。

  Future<String> readText() async {
    Future<String> readAsString = File(r"/Users/phoenix/Downloads/test1").readAsString();
    return readAsString;
  }
  

  获取返回值：

  void main() {
    readText().then((value) {
      print("读取文本:$value");
    });
  }
  

• await

  await可以将原本异步的操作变成同步操作。

  Future<String> readText() async {
    // 如果不写await关键字，则readAsString()是一个异步操作。在第一个没有读取完之后就开始执行第二方法了。
    // Future<String> readAsString = File(r"/Users/phoenix/Downloads/test1").readAsString();
    // var readAsString2 = File(r"/Users/phoenix/Downloads/test1").readAsString();
  
    // await 等待future执行完成后再执行后续代码
    // 使用了await关键字，则这段代码就变成同步，会阻塞后续代码的执行
    var readAsString =
        await File(r"/Users/phoenix/Downloads/test1").readAsString();
    var readAsString2 =
        await File(r"/Users/phoenix/Downloads/test1").readAsString();
    return readAsString + readAsString2;
  }
  
  void main() {
    readText().then((value) {
      print("读取文本:$value");
    });
  }