isolate
起因
最近看了一点isolate的东西, 自己写了一个例子.
普通的的 consummer-producer例子是只有前后两端的,实际上,会把前后两端再进行包装.
我这里这个例子,是把这个层级扩充,变成4层.
消费者 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()的时候,产生的乱码.