FLUTTER 中 Isolate 的一個例子.

isolate

原由

最近看了一點isolate的東西, 本身寫了一個例子.
普通的的 consummer-producer例子是隻有先後兩端的,實際上,會把先後兩端再進行包裝.
我這裏這個例子,是把這個層級擴充,變成4層.

graph LR A((消費者))--定單--> B((小商店))--定單-->C((批發商))--定單-->D((加工廠)) D--運送-->C--運送-->B--運送-->A

消費者 consummer 是一個Future,發送請求,等待回覆, 不關心中間過程.
小商店 seller 是主線程的處理函數,響應consummer的請求,並把請求發送給次線程的處理函數,
批發商 wholesale 是次線程的任務分發函數,能夠把任務按同步的方式分發,也能夠按異步的方式分發,個人例子是異步方式.
加工廠 producer 是執行任務的api, future. 在次線程執行.

同時,也要定義好先後通信的信息的格式.

代碼

dart代碼,flutter中的代碼就不放了.

import 'dart:isolate';

typedef FutureApiRets = Future<ApiRets> Function(ApiArgs args);

//api的參數的形式
class ApiArgs {
  List<dynamic> list;
  Map<dynamic, dynamic> map;
  ApiArgs(this.list, this.map);
  @override
  String toString() {
    return {
      'ApiArgs': {'list': this.list, 'map': this.map}
    }.toString();
  }
}
//api的返回值的形式
class ApiRets {
  List<dynamic> list;
  Map<dynamic, dynamic> map;
  ApiRets(this.list, this.map);
  @override
  String toString() {
    return {
      'ApiRets': {'list': this.list, 'map': this.map}
    }.toString();
  }
}
//seller 到 wholesale 的消息的格式
class Msg<T> {
  T msg;
  SendPort sendPort;
  String apiName;
  ApiArgs apiArgs;
  Msg(this.msg, ReceivePort receivePort, {this.apiName = "", this.apiArgs}) : sendPort = receivePort.sendPort;
}
//封裝 seller,wholesale 到一個類裏,同時把線程的初始化和結束功能寫出來.
class IsolateTest {
  static SendPort sendPort;
  static Isolate isolate;
  static bool inited = false;
  //seller類.
  static Future<K> seller<T, K>(T msg, {String apiName, ApiArgs args}) async {
    print("seller    start++:msg->${msg.toString()}");
    var rec = ReceivePort();
    //與wholescale經過sendPort通信
    IsolateTest.sendPort.send(Msg<T>(msg, rec, apiName: apiName, apiArgs: args));
    print("seller    wait===:msg->${msg.toString()}");
    K ret = (await rec.first) as K;
    print("seller    end----:msg->${msg.toString()}");
    return ret;
  }
  //wholesale功能,執行在次線程裏
  static Future wholesale(Msg message) async {
    print('wholesale start');
    ApiTest().name = 'worker';
    ReceivePort port = ReceivePort();
    message.sendPort.send(port.sendPort);
    print('wholesale waiting');
    int count = 0;
    //不斷的接受數據
    await for (Msg msg in port) {
      SendPort reply = msg.sendPort;
      if (reply == null) continue;
      print("wholesale start++:msg->${msg.msg}:count->$count");
      count++;
      var ret;
      //根據api的名稱來調用
      if (msg.apiName != "") {
        var s = () async {
          var ret;
          try {
            FutureApiRets api = ApiTest().apiMap[msg.apiName];
            if (api != null) {
              ret = await api(msg.apiArgs);
            }
          } catch (e) {
            print(e.toString());
          }
          return ret;
        };
        s().then((value) {
          print("wholesale end----:msg->${msg.msg}:count->$count");
          reply.send(value);
        });
      } else {
        print("wholesale end----:msg->${msg.msg}:count->$count");
        reply.send(ret);
      }
    }
    print('wholesale over');
  }
  //建立次線程
  static Future createIsolate() async {
    if (!inited) {
      print('create isolate start');
      ReceivePort rec = ReceivePort();
      Msg msg = Msg('000', rec);
      isolate = await Isolate.spawn(wholesale, msg);
      sendPort = await rec.first;
      inited = true;
      print('create isolate finish');
    }
  }
  //結束次線程
  static closeIsolate() {
    print('close isolate');
    if (isolate != null) {
      isolate.kill(priority: Isolate.immediate);
      isolate = null;
      inited = false;
    }
  }
}
//次線程內,執行具體任務的api,封裝在一個類裏,
class ApiTest {
  static ApiTest _ins;
  factory ApiTest() => _ins ?? ApiTest._fac();
  ApiTest._fac() {
    this.name = '';
    //把api按名字作索引,是否能夠重載[]或nosuchmethod來改進?
    this.apiMap = {
      'apiA': this.apiA,
      'apiB': this.apiB,
      'apiC': this.apiC,
    };
  }
  String name;
  Map<String, FutureApiRets> apiMap;
  //apis
  Future<ApiRets> apiA(ApiArgs args) async {
    print("producer  apiA+++:name:[${this.name}],args:${args.toString()}");
    await Future.delayed(Duration(seconds: 3));
    ApiRets ret = ApiRets(["apiA"], {});
    print("producer  apiA---:name:[${this.name}],args:${ret.toString()}");
    return ret;
  }

  Future<ApiRets> apiB(ApiArgs args) async {
    print("producer  apiB+++:name:[${this.name}],args:${args.toString()}");
    await Future.delayed(Duration(seconds: 4));
    ApiRets ret = ApiRets(["apiB"], {});
    print("producer  apiB---:name:[${this.name}],args:${ret.toString()}");
    return ret;
  }

  Future<ApiRets> apiC(ApiArgs args) async {
    print("producer  apiC+++:name:[${this.name}],args:${args.toString()}");
    await Future.delayed(Duration(seconds: 5));
    ApiRets ret = ApiRets(["apiC"], {});
    print("producer  apiC---:name:[${this.name}],args:${ret.toString()}");
    return ret;
  }
}
//用戶的調用測試
Future consumer(int count, String name, String apiName) async {
  ApiArgs args = ApiArgs([name, count], {});
  print('consumer  start++:$name,$count,${args.toString()}');
  ApiRets ret = await IsolateTest.seller<String, ApiRets>(name, apiName: apiName, args: args);
  print('consumer  end----:$name,$count,${ret.toString()}');
}

main(List<String> args) async {
  await IsolateTest.createIsolate();
  await Future.wait<dynamic>([
    consumer(1, 'aaa', 'apiA'),
    consumer(2, 'aaa', 'apiB'),
    consumer(3, 'aaa', 'apiC'),
    consumer(4, 'bbb', 'apiA').then((value) => consumer(5, "ccc", 'apic')),
  ]).whenComplete(() async {
    await IsolateTest.closeIsolate();
  });
}

運行結果

create isolate start
wholesale start
wholesale waiting
create isolate finish
consumer  start++:aaa,1,{ApiArgs: {list: [aaa, 1], map: {}}}
seller    start++:msg->aaa
seller    wait===:msg->aaa
consumer  start++:aaa,2,{ApiArgs: {list: [aaa, 2], map: {}}}
seller    start++:msg->aaa
seller    wait===:msg->aaa
consumer  start++:aaa,3,{ApiArgs: {list: [aaa, 3], map: {}}}
seller    start++:msg->aaa
seller    wait===:msg->aaa
consumer  start++:bbb,4,{ApiArgs: {list: [bbb, 4], map: {}}}
seller    start++:msg->bbb
seller    wait===:msg->bbb
wholesale start++:msg->aaa:count->0
producer  apiA+++:name:[],args:{ApiArgs: {list: [aaa, 1], map: {}}}
wholesale start++:msg->aaa:count->1
producer  apiB+++:name:[],args:{ApiArgs: {list: [aaa, 2], map: {}}}
wholesale start++:msg->aaa:count->2
producer  apiC+++:name:[],args:{ApiArgs: {list: [aaa, 3], map: {}}}
wholesale start++:msg->bbb:count->3
producer  apiA+++:name:[],args:{ApiArgs: {list: [bbb, 4], map: {}}}
producer  apiA---:name:[],args:{ApiRets: {list: [apiA], map: {}}}
wholesale end----:msg->aaa:count->4
seller    end----:msg->aaa
consumer  end----:aaa,1,{ApiRets: {list: [apiA], map: {}}}
producer  apiA---:name:[],args:{ApiRets: {list: [apiA], map: {}}}
wholesale end----:msg->bbb:count->4
seller    end----:msg->bbb
consumer  end----:bbb,4,{ApiRets: {list: [apiA], map: {}}}
consumer  start++:ccc,5,{ApiArgs: {list: [ccc, 5], map: {}}}
seller    start++:msg->ccc
seller    wait===:msg->cccwholesale start++:msg->ccc:count->4

wholesale end----:msg->ccc:count->5
seller    end----:msg->ccc
consumer  end----:ccc,5,null
producer  apiB---:name:[],args:{ApiRets: {list: [apiB], map: {}}}
wholesale end----:msg->aaa:count->5
seller    end----:msg->aaa
consumer  end----:aaa,2,{ApiRets: {list: [apiB], map: {}}}
producer  apiC---:name:[],args:{ApiRets: {list: [apiC], map: {}}}
wholesale end----:msg->aaa:count->5
seller    end----:msg->aaa
consumer  end----:aaa,3,{ApiRets: {list: [apiC], map: {}}}
close isolate

注意到中間 有部分的顯示有問題.

seller start++:msg->ccc
seller wait===:msg->cccwholesale start++:msg->ccc:count->4

wholesale end----:msg->ccc:count->5

這裏, 我分析是主次線程同時調用print()的時候,產生的亂碼.