isolate应用和封装-一一网

上一章，分享了Dart异步和isolate的基本知识。这一章，分享isolate的应用。

创建Isolate

基本方法
Flutter应用中的代码默认都跑在root isloate 中，尽管是单线程但已经足够处理各类异步任务。

当有计算密集型的耗时任务时，就需要创建新的Isolate来进行耗时计算来避免阻塞root isloate。由于不同Isolate之间内存隔离，要通信就得通过 ReceivePort与SendPort 来实现。

使用Isolate.spawn来创建新的Isolate。看下函数签名

external static Future<Isolate> spawn<T>(
  void entryPoint(T message), T message,
  {bool paused = false,
  bool errorsAreFatal = true,
  SendPort? onExit,
  SendPort? onError,
  @Since("2.3") String? debugName});
复制代码

创建新的Isolate最主要的就以下两步:

1、使用顶层函数或静态方法定义计算任务

// 耗时计算部分
int fibonacci(int n) {
  return n < 2 ? n : fibonacci(n - 2) + fibonacci(n - 1);
}

// 1. 计算任务
void task1(int start) {
  DateTime startTime = DateTime.now();
  int result = fibonacci(start);
  DateTime endTime = DateTime.now();
  print("计算耗时：${endTime.difference(startTime)}  结果：${result.toString()}");
}

void main() {
  task1(50);
}

// 输出：计算耗时：0:00:48.608656  结果：12586269025
复制代码

上面的计算要耗时48秒，必须要在新的Isolate中做计算。

最简单的 compute

通常网络返回 json ，我们需要解析成实体 bean ，如果 json 十分庞大，耗时较多，就卡顿了。所以需要放在isolate里处理。

import 'dart:convert';

main(List<String> args) {
  String jsonString = '''{ "id":"123", "name":"张三", "score" : 95}''';
  Student student = parseJson(jsonString);
  print(student.name);
}

Student parseJson(String json) {
  Map<String, dynamic> map = jsonDecode(json);
  return Student.fromJson(map);
}

class Student {
  String id;
  String name;
  int score;
  Student({this.id, this.name, this.score});
  factory Student.fromJson(Map parsedJson) {
    return Student(id: parsedJson['id'], name: parsedJson['name'], score: parsedJson['score']);
  }
}
复制代码

我们把上面代码放入isolate中执行：

Future<Student> loadStudent(String json) {
  return compute(parseJson, json);
}

Student parseJson(String json) {
  Map<String, dynamic> map = jsonDecode(json);
  return Student.fromJson(map);
}
复制代码

compute是 Flutter 的 api ，帮我们封装了 isolate，使用十分简单，但是也有局限性，它没有办法多次返回结果，也没有办法持续性的传值计算，每次调用，相当于新建一个隔离，如果同时调用过多的话反而会多次开辟内存。在某些业务下，我们可以使用compute，但是在另外一些业务下，我们只能使用dart提供的 isolate了。

单向通信

定义可以接收宿主的SendPort的耗时任务，在计算完成后通过send方法将结果发回宿主Isolate。

我们把上面的代码利用isolate实现一遍：

import 'dart:convert';
import 'dart:isolate';

main(List<String> args) async {
  await start();
}

Isolate isolate;

start() async {
  //创建接收端口，用来接收子线程消息
  ReceivePort receivePort = ReceivePort();

  //创建并发Isolate，并传入主线程发送端口
  isolate = await Isolate.spawn(entryPoint, receivePort.sendPort);
  //监听子线程消息
  receivePort.listen((data) {
    print('Data：$data');
  });
}

//并发Isolate
entryPoint(SendPort sendPort) {
  String jsonString = '''{ "id":"123", "name":"张三", "score" : 95}''';
  Student student = parseJson(jsonString);
  sendPort.send(student);
}

Student parseJson(String json) {
  Map<String, dynamic> map = jsonDecode(json);
  return Student.fromJson(map);
}

class Student {
  String id;
  String name;
  int score;
  Student({this.id, this.name, this.score});
  factory Student.fromJson(Map parsedJson) {
    return Student(id: parsedJson['id'], name: parsedJson['name'], score: parsedJson['score']);
  }
}
复制代码

双向通信

子Isolate向宿主Isolate发送数据是通过，持有宿主中的SendPort来实现的。

通过上面的改造，将宿主中的SendPort传递给子Isolate，在计算出结果后发回给宿主Isolate，宿主Isolate通过ReceivePort设置监听来处理结果。就完成了子Isolate向宿主Isolate发送数据。

但宿主Isolate如何向子Isolate发送数据呢？

子Isolate向宿主Isolate发送数据是通过，持有宿主中的SendPort来实现的。那么要实现宿主Isolate向子Isolate发送数据，宿主中也得持有子Isolate中的SendPort才行。最简单的方案就是在子Isolate中创建subSendPort并传递回宿主。

import 'dart:isolate';

main(List<String> args) async {
  await start();
  await Future.delayed(Duration(seconds: 1), () {
    threadPort.send('我来自主线程');
    print('1');
  });
  await Future.delayed(Duration(seconds: 1), () {
    threadPort.send('我也来自主线程');
    print('2');
  });
  await Future.delayed(Duration(seconds: 1), () {
    threadPort.send('end');
    print('3');
  });
}

Isolate isolate;
//子线程发送端口
SendPort threadPort;
start() async {
  //创建主线程接收端口，用来接收子线程消息
  ReceivePort receivePort = ReceivePort();

  //创建并发Isolate，并传入主线程发送端口
  isolate = await Isolate.spawn(entryPoint, receivePort.sendPort);
  //监听子线程消息
  receivePort.listen((data) {
    print('主线程收到来自子线程的消息$data');
    if (data is SendPort) {
      threadPort = data;
    }
  });
}

//并发Isolate
entryPoint(dynamic message) {
  //创建子线程接收端口，用来接收主线程消息
  ReceivePort receivePort = ReceivePort();

  SendPort sendPort;
  print('==entryPoint==$message');
  if (message is SendPort) {
    sendPort = message;
    print('子线程开启');
    sendPort.send(receivePort.sendPort);
    //监听子线程消息
    receivePort.listen((data) {
      print('子线程收到来自主线程的消息$data');
      assert(data is String);
      if (data == 'end') {
        isolate?.kill();
        isolate = null;
        print('子线程结束');
        return;
      }
    });
    return;
  }
}
复制代码

最终输出结果为：

==entryPoint==SendPort
子线程开启
主线程收到来自子线程的消息SendPort
1
子线程收到来自主线程的消息我来自主线程
2
子线程收到来自主线程的消息我也来自主线程
3
子线程收到来自主线程的消息end
子线程结束
复制代码

双向通信比较复杂，所以我们需要封装下，通过 api 让外部调用：

import 'dart:async';
import 'dart:isolate';

main(List<String> args) async {
  var worker = Worker();
  worker.reuqest('发送消息1').then((data) {
    print('子线程处理后的消息:$data');
  });

  Future.delayed(Duration(seconds: 2), () {
    worker.reuqest('发送消息2').then((data) {
      print('子线程处理后的消息:$data');
    });
  });
}

class Worker {
  SendPort _sendPort;
  Isolate _isolate;
  final _isolateReady = Completer<void>();
  final Map<Capability, Completer> _completers = {};

  Worker() {
    init();
  }

  void dispose() {
    _isolate.kill();
  }

  Future reuqest(dynamic message) async {
    await _isolateReady.future;
    final completer = new Completer();
    final requestId = new Capability();
    _completers[requestId] = completer;
    _sendPort.send(new _Request(requestId, message));
    return completer.future;
  }

  Future<void> init() async {
    final receivePort = ReceivePort();
    final errorPort = ReceivePort();
    errorPort.listen(print);
    receivePort.listen(_handleMessage);
    _isolate = await Isolate.spawn(
      _isolateEntry,
      receivePort.sendPort,
      onError: errorPort.sendPort,
    );
  }

  void _handleMessage(message) {
    if (message is SendPort) {
      _sendPort = message;
      _isolateReady.complete();
      return;
    }
    if (message is _Response) {
      final completer = _completers[message.requestId];
      if (completer == null) {
        print("Invalid request ID received.");
      } else if (message.success) {
        completer.complete(message.message);
      } else {
        completer.completeError(message.message);
      }
      return;
    }
    throw UnimplementedError("Undefined behavior for message: $message");
  }

  static void _isolateEntry(dynamic message) {
    SendPort sendPort;
    final receivePort = ReceivePort();

    receivePort.listen((dynamic message) async {
      if (message is _Request) {
        print('子线程收到：${message.message}');
        sendPort.send(_Response.ok(message.requestId, '处理后的消息'));
        return;
      }
    });

    if (message is SendPort) {
      sendPort = message;
      sendPort.send(receivePort.sendPort);
      return;
    }
  }
}

class _Request {
  /// The ID of the request so the response may be associated to the request's future completer.
  final Capability requestId;

  /// The actual message of the request.
  final dynamic message;

  const _Request(this.requestId, this.message);
}

class _Response {
  /// The ID of the request this response is meant to.
  final Capability requestId;

  /// Indicates if the request succeeded.
  final bool success;

  /// If [success] is true, holds the response message.
  /// Otherwise, holds the error that occured.
  final dynamic message;

  const _Response.ok(this.requestId, this.message) : success = true;

  const _Response.error(this.requestId, this.message) : success = false;
}
复制代码

感谢:

sg.jianshu.io/p/a4a871995…

juejin.cn/post/696136…

文章版权归作者所有，未经允许请勿转载。

THE END

IOS