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

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

如果你熟悉 JavaScript 的 Promise 模式的話，發起一個異步http請求，你可以這樣寫：

<code>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);
});
/<code>

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

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

<code>Future<response> respFuture = http.get('https://example.com'); //發起請求
respFuture.then((response) { //成功，匿名函數
  if (response.statusCode == 200) { 

    var data = reponse.data;
  }
}).catchError((error) { //失敗
   handle(error);
});
/<response>/<code>

這種模式簡化和統一了異步的處理，即便沒有系統學習過併發編程的同學，也可以拋開復雜的多線程，開箱即用。

Future

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

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

completed狀態也有兩種：一種是代表操作成功，返回結果；另一種代表操作失敗，返回錯誤。

我們來看一個例子：

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

void main() {
  fetchUserOrder().then((result){print(result)})
  print('Fetching user order...');
}
/<string>/<code>

通過then來回調成功結果，main會先於Future裡面的操作，輸出結果：

<code>Fetching user order...
Large Latte
/<code>

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

Future同名構造器是factory Future(FutureOr computation())，它的函數參數返回值為FutureOr類型，我們發現還有很多Future中的方法比如Future.then、Future.microtask的參數類型也是FutureOr，看來有必要了解一下這個對象。

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

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

你可以把它理解為受限制的dynamic類型，因為它只能接受Future或者T類型的值：

<code>FutureOr hello(){} 


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

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

在 Dart 的最佳實踐裡面明確指出：請避免聲明函數返回類型為FutureOr。

如果調用下面的函數，除非進入源代碼，否則無法知道返回值的類型究竟是int 還是Future：

<code>FutureOr triple(FutureOr value) async => (await value) * 3;
/<code>

正確的寫法：

<code>Future triple(FutureOr value) async => (await value) * 3;
 
/<code>

稍微交代了下FutureOr，我們繼續研究Future。

如果Future內的函數執行發生異常，可以通過Future.catchError來處理異常：

<code>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...');
}
/<void>/<code>

輸出結果：

<code>Fetching user order...
Exception: Logout failed: user ID is invalid
/<code>

Future支持鏈式調用：

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

void main() {
   fetchUserOrder().then((result) => result + 'BBB')
     .then((result) => result + 'CCC')
     .then((result){print(result);});
}
/<string>/<code>

輸出結果：

<code>AAABBBCCC
/<code>

async 和 await

想象一個這樣的場景：

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

接口定義：

<code>Future<string> login(String name,String password){
  //登錄
}
Future<user> fetchUserInfo(String token){
  //獲取用戶信息
}
Future saveUserInfo(User user){
  // 緩存用戶信息
}
/<user>/<string>/<code>

用Future大概可以這樣寫：

<code>login('name','password').then((token) => fetchUserInfo(token))
  .then((user) => saveUserInfo(user));
/<code>

換成async 和await 則可以這樣：

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

聲明瞭async 的函數，返回值是必須是Future對象。即便你在async函數里面直接返回T類型數據，編譯器會自動幫你包裝成Future

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

<code>void doLogin() async {
  try {
    var token = await login('name','password');
    var user = await fetchUserInfo(token);
    await saveUserInfo(user);
  } catch (err) {
    print('Caught error: $err');
  }
}
/<code>

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

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

送多一顆語法糖給你：

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

等價於：

Future<string> getUserInfo() async {
  return Future.value('aaa'); 

}
/<string>/<string>/<code>

Dart異步原理

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

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

I/O 模型

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

<code>int count = io.read(buffer); //阻塞等待
/<code>

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

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

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

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

如果進行IO的時候不用阻塞，那情況就不一樣了：

<code>while(true){
  for(io in io_array){
      status = io.read(buffer);// 不管有沒有數據都立即返回
      if(status == OK){
       
      }
  }
}
/<code>

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

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

<code>while(true){
    //如果其中一路IO有數據返回，則立即返回；如果一直沒有，最多等待不超過timeout時間 

    status = select(io_array, timeout); 
    if(status  == OK){
      for(io in io_array){
          io.read() //立即返回，數據都準備好了
      }
    }
}
/<code>

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

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

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

答案就是異步IO了：

<code>io.async_read((data) => {
  // dosomething
});
/<code>

通過異步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當做是異步的就行了。

Dart 源碼中的 epoll_wait

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

<code>Future<response> respFuture = http.get('https://example.com'); //發起請求
/<response>/<code>

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

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

事件循環 （Event Loop）

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

<code>while(true){
   event = event_queue.first() //取出事件
   handleEvent(event) //處理事件
   drop(event) //從隊列中移除
}
/<code>

比如下面這段代碼：

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

當你點擊屏幕上按鈕時，會產生一個事件，這個事件會放入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內部的任務，適合用來不會特別耗時或緊急的任務，微任務隊列的處理優先級比事件隊列的高，如果微任務處理比較耗時，會導致事件堆積，應用響應緩慢。

isolate event loop

你可以通過Future.microtask 來向isolate提交一個微任務：

<code>import 'dart:async';

main() { 

  new Future(() => print('beautiful'));
  Future.microtask(() => print('hi'));
}
/<code>

輸出：

<code>hi
beautiful
/<code>

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

多線程

注：以下當我們提到isolate的時候，你可以把它等同於線程，但我們知道它不僅僅是一個線程。

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

在Dart中，你可以通過Isolate.spawn 來創建一個新的isolate：

<code>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)); //主線程阻塞等待
}
/<code>

輸出：

<code>Hello, Im from new isolate!
/<code>

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

如果主isolate想接收子isolate的消息，可以在主isolate創建一個ReceivePort對象，並把對應的receivePort.sendPort作為新isolate入口函數參數傳入，然後通過ReceivePort綁定SendPort對象給主isolate發送消息：

<code>//新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);
  });
}
/<code>

輸出：

<code>hello, Im from new isolate!
/<code>

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

<code>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');
} 

/<sendport>/<sendport>/<code>

輸出：

<code>[mainToIsolateStream] This is from main isolatemain end
[isolateToMainStream] This is from new isolate
/<code>

在 Flutter 中，你還可以通過一個簡化版的compute函數啟動一個新的isolate。

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

<code>List<photo> parsePhotos(String responseBody) {
  final parsed = json.decode(responseBody).cast/<photo>/<code>

啟動一個新的isolate：

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

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

isolate 消息傳遞示意圖

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

1. Dart asynchronous programming: isolate and event loops
2. The Event Loop and Dart
3. Node.js 的異步 I/O 實現
4. Dart Isolate 2-Way Communication
5. 徹底搞懂Dart異步

關於AgileStudio

我們是一支由資深獨立開發者和設計師組成的團隊，成員均有紮實的技術實力和多年的產品設計開發經驗，提供可信賴的軟件定製服務。

閱讀更多 AgileStudio 的文章

關鍵字: Sync 異步 數據庫