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()的时候,产生的乱码.