深入探索 Dart 的併發與異步處理：從 Isolate 到 Event Loop 的全面指南 | Future、Stream

透過 isolate#spawn 方法可以開啟一個新的執行序

import 'dart:isolate';

int? data;     // int 是類，所有 default 是 null

void main() {
  data = 10;

  // 開新線程，spawn 是泛型
  // 原型 : external static Future<Isolate> spawn<T> (方法, ,{...})
  Isolate.spawn(entryPoint, "Alien");

  print("Finish main isolate, data: $data");   // 判斷是否會排隊進行任務  and  數據是否隔離
}

void entryPoint(String message) { // 可以省去 (message)
  print("In child isolate, Params: $message, data: $data");   // 證明數據是隔離的
}

從執行結果來看我們可以知道

A. spawn 方法確實可以開啟一個新執行序，因為主執行緒正常會先結束，而開啟的執行序在之後才結束

B. isolate#spawn 方法開啟的執行序是預設不守護主執行緒的，因為主執行緒結束後，透過 spawn 開啟的執行緒也不會結束

要正名資源的隔離也相當容易，讓不同的執行序持有相同資源的引用，如果資源被改變則是資源共享，而資源沒有被改變則是資源不共享！（也就是資源被隔離）

import 'dart:isolate';

int i = -1;

void main() {
  i = 20;

  Isolate.spawn(entryPoint, "Test");    // Java 可想成 new Thread

  print("Main $i");
}

void entryPoint(String msg) {
  print("Isolate $i");
}

從結果可以看到，兩者所取得的資源並不相同，Isolate.spawn 開啟的執行序仍舊是保持初始化的 -1（因為開啟執行緒後執行的 entryPoint 函數，它是取全域的 i 資源）

● 如果是資源共享的情況下，會變怎樣？

資源 i 如果是共享的，那代表主執行緒內將 i 修改為 20，而後 Isolate.spawn 開啟的執行序也會讀取到 i 為 20

import 'dart:isolate';

void main() {

  ReceivePort receivePort = ReceivePort();

  // 創建 isolate 執行緒
  Isolate.spawn<SendPort>(isolateFunc, receivePort.sendPort);

  // receive 監聽
  // 原型: StreamSubscription listen(void onData(var message), {...});
  receivePort.listen(mainListenIsolate);    // main 監聽 子 isolate 訊息

  print("Main~~~~~~ finish");      // Main 並不會等待 isolate !
}

// main isolate 監聽子 isolate

void mainListenIsolate(var message) {
  if(message is SendPort) {

    print("I am Main, I get your send Port");
    message.send("Main get");      // 2.

  } else {
    // 接收子 isolate 訊息
    print("Main get Message from isolate: $message");
  }
}

// isolate 執行序

void isolateFunc(SendPort sendPort) {
  print('isolate created');
  
  // 創建自己的接收器
  ReceivePort receivePort = ReceivePort();
  // 透過主執行緒給予的 SendPort 物件發送把 isolate 內的發送器（`SendPort`）發送給主執行緒
  sendPort.send(receivePort.sendPort);    

  // 監聽 Main
  receivePort.listen((var message) {
    print("Child Isolate get Message from Main: $message");
  });

  sendPort.send("I am isolate");    // 透過主執行緒給予的 SendPort 物件發送資料給主執行緒
}

但是與 Dart 不同的是，Android 天生屬於多執行序

黃色區塊為子執行序，要注意 Main 函數會正常執行，並 併發執行 Event-Loop

每次 Event-loop 在執行 EventQueue 任務前，都會檢查 MicrotaskQueue 是否有需要執行的微任務，若 MicrotaskQueue 有任務要執行會先執行，之後才接續執行 EventQueue 任務

● Event-loop 的啟動：

Event-loop 在 Dart 應用中是自動啟動的，不需要等 main 完全執行完畢才開始運行… 實際上，Event-loop 是一個「持續運行的機制」，從應用啟動之後就開始運行

then 函數是在 Event-Loop 檢查 EventQueue 時才執行，只要 EventQueue 內有任務就會被執行（以下任務是 IO）

import 'dart:io';

void main() {

  new File(r"app_entry.dart").readAsString().then((value) {
    print("執行 Task");
    print(value);
  });

  print("read file finish");
}

這裡我們來驗證將任務插入至 MicrotaskQueue 是否會比起把任務插入 EventQueue 還要快被執行

插入任務至 MicrotaskQueue 需要使用 Future.microtask 函數

import 'dart:io';

void main() {

  new File(r"app_entry.dart").readAsString().then((value) {
    print("執行 Task");
  });

  // 原型 : factory Future.microtask(FutureOr<T> computation()) {
  Future.microtask(() {
    print("執行 micro Task");
  });


  print("Main finish");
}

如下結果，我們可以看到 微任務的執行會在一般任務之前！

● Dart 事件循環：當主執行緒被阻塞時（例如透過 sleep），事件循環無法繼續，非同步任務的回呼也無法執行

● Kotlin 協程：Kotlin 的協程能夠在任務之間智慧切換，避免主執行緒阻塞，能夠繼續處理其他任務

Event-loop 是用來掃描其內部的兩個 Queue 事件的機制，而我們撰寫的 異步任務就會放置到 Queue 中

那放置在兩個 Queue 的任務何時被執行呢？

Dart 不會另外起一執行序來執行，仍舊保值單一執行序執行，而這個執行是採用「協作式」的方式執行… 也就是說「異步任務會在主執行序空閒時被執行」

import 'dart:async';

void main() {
  Future<String> myFuture = Future.delayed(
      Duration(seconds: 2), 
      () => 'Hello from Future!');

  Stream<String> streamFromFuture = Stream.fromFuture(myFuture);

  streamFromFuture.listen(
    (data) {
      print('Data: $data');
    },
    onError: (error) {
      print('Error: $error');
    },
    onDone: () {
      print('Stream closed.');
    },
  );
}

import 'dart:async';

void main() {
  Future<String> future1 = Future.delayed(Duration(seconds: 1), () => 'First Future');
  Future<String> future2 = Future.delayed(Duration(seconds: 2), () => 'Second Future');
  Future<String> future3 = Future.delayed(Duration(seconds: 3), () => 'Third Future');

  Stream<String> streamFromFutures = Stream.fromFutures([future1, future2, future3]);

  streamFromFutures.listen(
    (data) {
      print('Data: $data');
    },
    onError: (error) {
      print('Error: $error');
    },
    onDone: () {
      print('Stream closed.');
    },
  );
}

import 'dart:async';

void main() {
  List<int> myIterable = [1, 2, 3, 4, 5];

  Stream<int> streamFromIterable = Stream.fromIterable(myIterable);

  streamFromIterable.listen(
    (data) {
      print('Data: $data');
    },
    onError: (error) {
      print('Error: $error');
    },
    onDone: () {
      print('Stream closed.');
    },
  );
}

void main() {

  var stream = Stream.fromIterable([1, 2, 3]);
  // 3 秒後添加訂閱者，Timer 延遲 (時間，callback)
  Timer(const Duration(seconds: 3),
          () => stream.listen((Object o) => print("延遲三秒: $o")));

}

void main() {

  //創建一個 StreamController
  var streamController = StreamController.broadcast();  // factory 構造器
  // 在以後微任務中獲得事件
  streamController.add("Hello");  // 無法接收，因為尚未有 listener
  //訂閱事件
  streamController.stream.listen((i){
    print("StreamController broadcast: $i");
  });
  streamController.add("World");  // 可以被接收，因為已有 listener
  // 記得關閉流 !!!
  streamController.close();

}

如下圖所見，在 listen 之前的 Hello 數據是無法被 StreamController 監聽到的，只有在 listen 之後的 World 可被監聽到

void main() async {
  // 使用 countStream 函數生成一個 Stream
  await for (int value in countStream(5)) {
    print('Received: $value');
  }

  print('Stream processing completed.');
}

void main() {
  // 使用 countStream 函數生成一個 Stream
  final stream = countStream(5);

  // 訂閱這個 Stream 並監聽數據
  stream.listen((value) {
      print('Received: $value');
    },
    onDone: () {
      print('Stream processing completed.');
    },
  );
}

import 'dart:async';

class EventBus {

  static EventBus _instance;
  static StreamController _streamController;

  EventBus._internal() {
    _streamController = StreamController.broadcast();
  }

  factory EventBus.getDefault() {
    // 單例模式
    return _instance ??= EventBus._internal();
  }

  StreamSubscription<T> register<T>(void onData(T event)) {
    if(T == dynamic) {  // 未指定類型
      return _streamController.stream.listen(onData);
    } else {
      // 傳送指定類型
      return _streamController.stream.where((type) => type is T)  // 判斷
          .cast<T>()        // 強制轉型
          .listen(onData);
    }
  }

  // 傳送廣播資訊，多廣播
  void post<T>(T msg) {
    _streamController.add(msg);
  }

  // 關閉廣播流
  void close() {
    _streamController.close();
  }

}

import 'eventBus.dart';

void main() {

  EventBus.getDefault().register((event) {
    print("Receive All Message, from EventBus: $event");
  });

  EventBus.getDefault().register<String> ((event) {
    print("--- Receive String Message, from EventBus: $event");
  });

  EventBus.getDefault().register<int>((event) {
    print("--- --- Receive int Message, from EventBus: $event");
  });

  EventBus.getDefault().register<double>((event) {
    print("--- --- --- Receive double Message, from EventBus: $event");
  });


  EventBus.getDefault().post(1111);
  EventBus.getDefault().post("Hello World");
  EventBus.getDefault().post(765.123);

  EventBus.getDefault().close();
}

void noSync() {
  // 讀取順序不同，File1(linux-5.6.4.tar.xz) 較大的會讀得比較慢，File2(build.gradle) 會先讀完
  new File(r"D:\Linux Code\linux-5.6.4.tar.xz").readAsBytes().then((value) => print("file1 read finish"));
  new File(r"D:\mingw64\build.gradle").readAsString().then((value) => print("file2 read finish"));
}

import 'dart:io';

// 多重回調，容易混亂
void oldStyle() {
  new File(r"D:\Linux Code\linux-5.6.4.tar.xz").readAsBytes().then((value) {
    print("then --- file1 read finish");

    new File(r"D:\mingw64\build.gradle").readAsString().then((value) {
      print("then --- file2 read finish");
    });
  });
}

import 'dart:io';

Future orderReadFile() async {
  // await 異步中的同步
  await new File(r"D:\Linux Code\linux-5.6.4.tar.xz").readAsBytes().then((value) => print("async --- file1 read finish"));

  var file2 = await new File(r"D:\mingw64\build.gradle").readAsString().then((value) => print("async --- file1 read finish"));

  return file2;
}