Dart 異步編程

時間 2020-01-16

標籤 dart 異步編程简体版

原文原文鏈接

本文是【從零開始，一塊兒學習開發個 Flutter App 吧】路上的第 2 篇文章。java

本文將解決上一篇留下的問題： Dart 中是如何進行異步處理的？咱們首先簡單介紹了 Dart 中經常使用的異步處理 Future、sync和await ；第二部分試圖分析Dart做爲單線程語言的異步實現原理，進一步介紹IO模型和事件循環模型；最後介紹如何在 Dart 實現多線程以線程的相互通訊。node

若是你熟悉 JavaScript 的 Promise 模式的話，發起一個異步http請求，你能夠這樣寫：linux

new Promise((resolve, reject) =>{
    // 發起請求
    const xhr = new XMLHttpRequest();
    xhr.open("GET", 'https://www.nowait.xin/');
    xhr.onload = () => resolve(xhr.responseText); 
    xhr.onerror = () => reject(xhr.statusText);
    xhr.send();
}).then((response) => { //成功
   console.log(response);
}).catch((error) => { // 失敗
   console.log(error);
});
複製代碼

Promise 定義了一種異步處理模式：do... success... or fail...。android

在 Dart 中，與之對應的是Future對象：web

Future<Response> respFuture = http.get('https://example.com'); //發起請求
respFuture.then((response) { //成功，匿名函數
  if (response.statusCode == 200) {
    var data = reponse.data;
  }
}).catchError((error) { //失敗
   handle(error);
});
複製代碼

這種模式簡化和統一了異步的處理，即使沒有系統學習過併發編程的同窗，也能夠拋開復雜的多線程，開箱即用。數據庫

Future

Future 對象封裝了Dart 的異步操做，它有未完成（uncompleted）和已完成（completed）兩種狀態。編程

在Dart中，全部涉及到IO的函數都封裝成Future對象返回，在你調用一個異步函數的時候，在結果或者錯誤返回以前，你獲得的是一個uncompleted狀態的Future。json

completed狀態也有兩種：一種是表明操做成功，返回結果；另外一種表明操做失敗，返回錯誤。緩存

咱們來看一個例子：bash

Future<String> fetchUserOrder() {
  //想象這是個耗時的數據庫操做
  return Future(() => 'Large Latte');
}

void main() {
  fetchUserOrder().then((result){print(result)})
  print('Fetching user order...');
}
複製代碼

經過then來回調成功結果，main會先於Future裏面的操做，輸出結果：

Fetching user order...
Large Latte
複製代碼

在上面的例子中，() => 'Large Latte')是一個匿名函數，=> 'Large Latte' 至關於 return 'Large Latte'。

Future同名構造器是factory Future(FutureOr<T> computation())，它的函數參數返回值爲FutureOr<T>類型，咱們發現還有不少Future中的方法好比Future.then、Future.microtask的參數類型也是FutureOr<T>，看來有必要了解一下這個對象。

FutureOr<T> 是個特殊的類型，它沒有類成員，不能實例化，也不能夠繼承，看來它極可能只是一個語法糖。

abstract class FutureOr<T> {
  // Private generative constructor, so that it is not subclassable, mixable, or
  // instantiable.
  FutureOr._() {
    throw new UnsupportedError("FutureOr can't be instantiated");
  }
}
複製代碼

你能夠把它理解爲受限制的dynamic類型，由於它只能接受Future<T>或者T類型的值：

FutureOr<int> hello(){}

void main(){
   FutureOr<int> a = 1; //OK
   FutureOr<int> b = Future.value(1); //OK
   FutureOr<int> aa = '1' //編譯錯誤

   int c = hello(); //ok
   Future<int> cc = hello(); //ok
   String s = hello(); //編譯錯誤
}
複製代碼

在 Dart 的最佳實踐裏面明確指出：請避免聲明函數返回類型爲FutureOr<T>。

若是調用下面的函數，除非進入源代碼，不然沒法知道返回值的類型到底是int 仍是Future<int>：

FutureOr<int> triple(FutureOr<int> value) async => (await value) * 3;
複製代碼

正確的寫法：

Future<int> triple(FutureOr<int> value) async => (await value) * 3;
複製代碼

稍微交代了下FutureOr<T>，咱們繼續研究Future。

若是Future內的函數執行發生異常，能夠經過Future.catchError來處理異常：

Future<void> fetchUserOrder() {
  return Future.delayed(Duration(seconds: 3), () => throw Exception('Logout failed: user ID is invalid'));
}

void main() {
  fetchUserOrder().catchError((err, s){print(err);});
  print('Fetching user order...');
}
複製代碼

輸出結果：

Fetching user order...
Exception: Logout failed: user ID is invalid
複製代碼

Future支持鏈式調用：

Future<String> fetchUserOrder() {
  return Future(() => 'AAA');
}

void main() {
   fetchUserOrder().then((result) => result + 'BBB')
     .then((result) => result + 'CCC')
     .then((result){print(result);});
}
複製代碼

輸出結果：

AAABBBCCC
複製代碼

async 和 await

想象一個這樣的場景：

先調用登陸接口；
根據登陸接口返回的token獲取用戶信息；
最後把用戶信息緩存到本機。

接口定義：

Future<String> login(String name,String password){
  //登陸
}
Future<User> fetchUserInfo(String token){
  //獲取用戶信息
}
Future saveUserInfo(User user){
  // 緩存用戶信息
}
複製代碼

用Future大概能夠這樣寫：

login('name','password').then((token) => fetchUserInfo(token))
  .then((user) => saveUserInfo(user));
複製代碼

換成async 和await 則能夠這樣：

void doLogin() async {
  String token = await login('name','password'); //await 必須在 async 函數體內
  User user = await fetchUserInfo(token);
  await saveUserInfo(user);
}
複製代碼

聲明瞭async 的函數，返回值是必須是Future對象。即使你在async函數裏面直接返回T類型數據，編譯器會自動幫你包裝成Future<T>類型的對象，若是是void函數，則返回Future<void>對象。在遇到await的時候，又會把Futrue類型拆包，又會原來的數據類型暴露出來，請注意，await 所在的函數必須添加async關鍵詞。

await的代碼發生異常，捕獲方式跟同步調用函數同樣：

void doLogin() async {
  try {
    var token = await login('name','password');
    var user = await fetchUserInfo(token);
    await saveUserInfo(user);
  } catch (err) {
    print('Caught error: $err');
  }
}
複製代碼

得益於async 和await 這對語法糖，你能夠用同步編程的思惟來處理異步編程，大大簡化了異步代碼的處理。

注：Dart 中很是多的語法糖，它提升了咱們的編程效率，但同時也會讓初學者容易感到迷惑。

送多一顆語法糖給你：

Future<String> getUserInfo() async {
  return 'aaa';
}

等價於：

Future<String> getUserInfo() async {
  return Future.value('aaa');
}
複製代碼

Dart異步原理

Dart 是一門單線程編程語言。對於平時用 Java 的同窗，首先可能會反應：那若是一個操做耗時特別長，不會一直卡住主線程嗎？好比Android，爲了避免阻塞UI主線程，咱們不得不經過另外的線程來發起耗時操做（網絡請求/訪問本地文件等），而後再經過Handler來和UI線程溝通。Dart 到底是如何作到的呢？

先給答案：異步 IO + 事件循環。下面具體分析。

I/O 模型

咱們先來看看阻塞IO是什麼樣的：

int count = io.read(buffer); //阻塞等待
複製代碼

注： IO 模型是操做系統層面的，這一小節的代碼都是僞代碼，只是爲了方便理解。

當相應線程調用了read以後，它就會一直在那裏等着結果返回，什麼也不幹，這是阻塞式的IO。

但咱們的應用程序常常是要同時處理好幾個IO的，即使一個簡單的手機App，同時發生的IO可能就有：用戶手勢（輸入），若干網絡請求(輸入輸出)，渲染結果到屏幕（輸出）；更不用說是服務端程序，成百上千個併發請求都是屢見不鮮。

有人說，這種狀況可使用多線程啊。這確實是個思路，但受制於CPU的實際併發數，每一個線程只能同時處理單個IO，性能限制仍是很大，並且還要處理不一樣線程之間的同步問題，程序的複雜度大大增長。

若是進行IO的時候不用阻塞，那狀況就不同了：

while(true){
  for(io in io_array){
      status = io.read(buffer);// 無論有沒有數據都當即返回
      if(status == OK){
       
      }
  }
}
複製代碼

有了非阻塞IO，經過輪詢的方式，咱們就能夠對多個IO進行同時處理了，但這樣也有一個明顯的缺點：在大部分狀況下，IO都是沒有內容的（CPU的速度遠高於IO速度），這樣就會致使CPU大部分時間在空轉，計算資源依然沒有很好獲得利用。

爲了進一步解決這個問題，人們設計了IO多路轉接（IO multiplexing），能夠對多個IO監聽和設置等待時間：

while(true){
    //若是其中一路IO有數據返回，則當即返回；若是一直沒有，最多等待不超過timeout時間
    status = select(io_array, timeout); 
    if(status  == OK){
      for(io in io_array){
          io.read() //當即返回，數據都準備好了
      }
    }
}
複製代碼

IO 多路轉接有多種實現，好比select、poll、epoll等，咱們不具體展開。

有了IO多路轉接，CPU資源利用效率又有了一個提高。

眼尖的同窗可能有發現，在上面的代碼中，線程依然是可能會阻塞在 select 上或者產生一些空轉的，有沒有一個更加完美的方案呢？

答案就是異步IO了：

io.async_read((data) => {
  // dosomething
});
複製代碼

經過異步IO，咱們就不用不停問操做系統：大家準備好數據了沒？而是一有數據系統就會經過消息或者回調的方式傳遞給咱們。這看起來很完美了，但不幸的是，不是全部的操做系統都很好地支持了這個特性，好比Linux的異步IO就存在各類缺陷，因此在具體的異步IO實現上，不少時候可能會折中考慮不一樣的IO模式，好比 Node.js 的背後的libeio庫，實質上採用線程池與阻塞 I/O 模擬出來的異步 I/O [1]。

Dart 在文檔中也提到是借鑑了 Node.js 、EventMachine, 和 Twisted 來實現的異步IO，咱們暫不深究它的內部實現（筆者在搜索了一下Dart VM的源碼，發如今android和linux上彷佛是經過epoll實現的），在Dart層，咱們只要把IO當作是異步的就好了。

咱們再回過頭來看看上面Future那段代碼：

Future<Response> respFuture = http.get('https://example.com'); //發起請求
複製代碼

如今你知道，這個網絡請求不是在主線程完成的，它實際上把這個工做丟給了運行時或者操做系統。這也是 Dart 做爲單進程語言，但進行IO操做卻不會阻塞主線程的緣由。

終於解決了Dart單線程進行IO也不會卡的疑問，但主線程如何和大量異步消息打交道呢？接下來咱們繼續討論Dart的事件循環機制（Event Loop）。

事件循環（Event Loop）

在Dart中，每一個線程都運行在一個叫作isolate的獨立環境中，它的內存不和其餘線程共享，它在不停幹一件事情：從事件隊列中取出事件並處理它。

while(true){
   event = event_queue.first() //取出事件
   handleEvent(event) //處理事件
   drop(event) //從隊列中移除
}
複製代碼

好比下面這段代碼：

RaisedButton(
  child: Text('click me');
  onPressed: (){ // 點擊事件 
     Future<Response> respFuture = http.get('https://example.com'); 
     respFuture.then((response){ // IO 返回事件
        if(response.statusCode == 200){
           print('success');
        }
     })
  }
)
複製代碼

當你點擊屏幕上按鈕時，會產生一個事件，這個事件會放入isolate的事件隊列中；接着你發起了一個網絡請求，也會產生一個事件，依次進入事件循環。

在線程比較空閒的時候，isolate還能夠去搞搞垃圾回收（GC），喝杯咖啡什麼的。

API層的Future、Stream、async 和 await 實際都是對事件循環在代碼層的抽象。結合事件循環，回到對Future對象的定義（An object representing a delayed computation.），就能夠這樣理解了：isolate大哥，我快遞一個代碼包裹給你，你拿到後打開這個盒子，並順序執行裏面的代碼。

事實上，isolate 裏面有兩個隊列，一個就是事件隊列（event queue），還有一個叫作微任務隊列（microtask queue）。

事件隊列：用來處理外部的事件，若是IO、點擊、繪製、計時器（timer）和不一樣 isolate 之間的消息事件等。

微任務隊列：處理來自於Dart內部的任務，適合用來不會特別耗時或緊急的任務，微任務隊列的處理優先級比事件隊列的高，若是微任務處理比較耗時，會致使事件堆積，應用響應緩慢。

你能夠經過Future.microtask 來向isolate提交一個微任務：

import 'dart:async';

main() {
  new Future(() => print('beautiful'));
  Future.microtask(() => print('hi'));
}
複製代碼

輸出：

hi
beautiful
複製代碼

總結一下事件循環的運行機制：當應用啓動後，它會建立一個isolate，啓動事件循環，按照FIFO的順序，優先處理微任務隊列，而後再處理事件隊列，如此反覆。

多線程

注：如下當咱們提到isolate的時候，你能夠把它等同於線程，但咱們知道它不只僅是一個線程。

得益於異步 IO + 事件循環，儘管Dart是單線程，通常的IO密集型App應用一般也能得到出色的性能表現。但對於一些計算量巨大的場景，好比圖片處理、反序列化、文件壓縮這些計算密集型的操做，只單靠一個線程就不夠用了。

在Dart中，你能夠經過Isolate.spawn 來建立一個新的isolate：

void newIsolate(String mainMessage){
  sleep(Duration(seconds: 3));
  print(mainMessage);
}

void main() {
  // 建立一個新的isolate，newIoslate
  Isolate.spawn(newIsolate, 'Hello, Im from new isolate!'); 
  sleep(Duration(seconds: 10)); //主線程阻塞等待
}
複製代碼

輸出：

Hello, Im from new isolate!
複製代碼

spawn 有兩個必傳參數，第一個是新isolate入口函數（entrypoint），第二個是這個入口函數的參數值（message）。

若是主isolate想接收子isolate的消息，能夠在主isolate建立一個ReceivePort對象，並把對應的receivePort.sendPort做爲新isolate入口函數參數傳入，而後經過ReceivePort綁定SendPort對象給主isolate發送消息：

//新isolate入口函數
void newIsolate(SendPort sendPort){
  sendPort.send("hello, Im from new isolate!");
}

void main() async{
  ReceivePort receivePort= ReceivePort();
  Isolate isolate = await Isolate.spawn(newIsolate, receivePort.sendPort);
  receivePort.listen((message){ //監遵從新isolate發送過來的消息
   
    print(message);
     
    // 再也不使用時，關閉管道
     receivePort.close();
     
    // 關閉isolate線程
     isolate?.kill(priority: Isolate.immediate);
  });
}
複製代碼

輸出：

hello, Im from new isolate!
複製代碼

上面咱們瞭解了主isolate是如何監聽來自子isolate的消息的，若是同時子isolate也想知道主isolate的一些狀態，那該如何處理呢？下面的代碼將提供一種雙向通訊的方式：

Future<SendPort> initIsolate() async {
  Completer completer = new Completer<SendPort>();
  ReceivePort isolateToMainStream = ReceivePort();

  //監聽來自子線程的消息
  isolateToMainStream.listen((data) {
    if (data is SendPort) {
      SendPort mainToIsolateStream = data;
      completer.complete(mainToIsolateStream);
    } else {
      print('[isolateToMainStream] $data');
    }
  });

  Isolate myIsolateInstance = await Isolate.spawn(newIsolate, isolateToMainStream.sendPort);
  //返回來自子isolate的sendPort
  return completer.future; 
}

void newIsolate(SendPort isolateToMainStream) {
  ReceivePort mainToIsolateStream = ReceivePort();
  //關鍵實現：把SendPort對象傳回給主isolate
  isolateToMainStream.send(mainToIsolateStream.sendPort);

  //監聽來自主isolate的消息
  mainToIsolateStream.listen((data) {
    print('[mainToIsolateStream] $data');
  });

  isolateToMainStream.send('This is from new isolate');
}

void main() async{
  SendPort mainToIsolate = await initIsolate();
  mainToIsolate.send('This is from main isolate');
}
複製代碼

輸出：

[mainToIsolateStream] This is from main isolatemain end
[isolateToMainStream] This is from new isolate
複製代碼

在 Flutter 中，你還能夠經過一個簡化版的compute函數啓動一個新的isolate。

好比在反序列化的場景中，直接在主isolate進行序列化：

List<Photo> parsePhotos(String responseBody) {
  final parsed = json.decode(responseBody).cast<Map<String, dynamic>>();

  return parsed.map<Photo>((json) => Photo.fromJson(json)).toList();
}

Future<List<Photo>> fetchPhotos(http.Client client) async {
  final response =
      await client.get('https://jsonplaceholder.typicode.com/photos');
  //直接在主isolate轉換
  return parsePhotos(response.body); 
}
複製代碼

啓動一個新的isolate：

Future<List<Photo>> fetchPhotos(http.Client client) async {
  final response =
      await client.get('https://jsonplaceholder.typicode.com/photos');
  // 使用compute函數，啓動一個新的isolate
  return compute(parsePhotos, response.body);
}
複製代碼

本示例的完整版：Parse JSON in the background

總結一下，當遇到計算密集型的耗時操做，你能夠開啓一個新的isolate來併發執行任務。不像咱們常規認識的多線程，不一樣的isolate之間不能共享內存，但經過ReceivePort和SendPort能夠構建不一樣isolate之間的消息通道，另外從別的isolate傳來的消息也是要通過事件循環的。