天天看點
傳回

Dart 異步程式設計Futureasync 和 awaitDart異步原理多線程參考資料

本文是【 從零開始，一起學習開發個 Flutter App 吧

】路上的第 2 篇文章。

本文将解決上一篇留下的問題： Dart 中是如何進行異步處理的？我們首先簡單介紹了 Dart 中常用的異步處理

Future

 、

sync

 和

await

；第二部分試圖分析Dart作為單線程語言的異步實作原理，進一步介紹IO模型和事件循環模型；最後介紹 如何在 Dart 實作多線程以線程的互相通信。

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

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...。

在 Dart 中，與之對應的是

Future

對象：

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

。

completed

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

我們來看一個例子：

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

想象一個這樣的場景：

  1. 先調用登入接口；
  2. 根據登入接口傳回的token擷取使用者資訊；
  3. 最後把使用者資訊緩存到本機。

接口定義：

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

Dart 異步程式設計Futureasync 和 awaitDart異步原理多線程參考資料

我們再回過頭來看看上面

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内部的任務，适合用來不會特别耗時或緊急的任務，微任務隊列的處理優先級比事件隊列的高，如果微任務處理比較耗時，會導緻事件堆積，應用響應緩慢。

Dart 異步程式設計Futureasync 和 awaitDart異步原理多線程參考資料

你可以通過

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
Dart 異步程式設計Futureasync 和 awaitDart異步原理多線程參考資料

總結一下，當遇到計算密集型的耗時操作，你可以開啟一個新的

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

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

json 緩存 Dart 前端開發 javascript java Linux 編譯器 android開發 資料格式 data程式設計 part程式設計 dom程式設計 dart搭建 dart連結
上一篇: 挑戰：微服務內建測試挑戰：微服務內建測試Consumer-Driven Contracts：合約測試架構Pact我們的例子：
下一篇: 馬來西亞電商企業PrestoMall抛棄甲骨文，轉投阿裡雲POLARDB成本節省40%

繼續閱讀