Isolate.dart阅读

OGeek|极客世界-中国程序员成长平台 › 门户 › 编程› 服务器相关›Dart 编码风格指南

原作者: [db:作者] 来自: [db:来源] 收藏邀请

它是一个独立的dart执行上下文,有自己独立的空间,不能被其他的isolate直接访问,只能访问内部的资源,
其它isolate如果需要访问,只能通过对应的端口访问
当通过spawning方法生成一个新的isolate,调用spawning方法的isolate将会获得一个新的isolate如果创建成功
isolate有自己独立的空间，单独运行这自己的event loop,和主isolate一样有自己独立的taskqueue.

方法介绍

class Isolate {
 //ping和kill的参数,常量标志位,代表立即执行
  static const int immediate = 0;
//ping和kill命令的参数,代表在一下一次事件之前执行
  static const int beforeNextEvent = 1;

//控制端口,用来发送控制类型的消息给isolate,举个不恰当的栗子，它是一个插头,isolate就像是插座，他们之间有着一一对应的关系，电流好比消息,这样应该比较容易理解了把，给插座充电(发送消息给isolate)
 
//一些特定类型的消息需要isolate的支持,比如发送暂停或者终止能力，就比如充电,电流太小或这太大，插座可能不支持,太小无法提供能量，太大会烧坏。这里其实指的是 controlport发送的消息 isolate处理不了，无法接收,消息就被无视了。
 
    //通过非空判定,以及它和controlport关联的是否为当前对应的isolate来决定controlport是否可以发送暂停消息给isolate
  final Capability pauseCapability;

  //通过非空判定,以及它和controlport关联的是否为当前对应的isolate来决定controlport是否可以发送终止消息给isolate
  final Capability terminateCapability;
 
//设置debug name便于调试,它在转换为c embedding API的时候会被mapping成 Dart_DebugName.
  external String get debugName;

 //创建一个受限制的`Isolate`,它有2个capability,如果不传将失去这个能力
  Isolate(this.controlPort, {this.pauseCapability, this.terminateCapability});

  //获取当前正在使用的Isolate
  external static Isolate get current;

  //Ioslate执行的包路径,如果没有设置它将返回为null
  external static Future<Uri> get packageConfig;

  // 创建一个新的Isolate, entry-point 是当前Isolate的参数,它是一个函数指针, 初始化之后 entrypoint函数将会首先在新建的Isolate内部调用,并message作为它的参数
 //这个方法必须在最高层级或
 //通常初始化的 message会包含一个SendPort，这样就能和外部的传入message的Isolate双向通信了。  
 //paused代表了当前isolate的创建之后的初始化状态,如果需要发送消息，确保它不是暂停状态
  external static Future<Isolate> spawn<T>(
      void entryPoint(T message), T message,
      {bool paused: false,
      bool errorsAreFatal,
      SendPort onExit,
      SendPort onError,
      @Since("2.3") String debugName});

 //从指定的package中初始化一个Isolate, 并调用它的main函数,同时监听它的错误回调
  external static Future<Isolate> spawnUri(
      Uri uri,
      List<String> args,
      var message,
      {bool paused: false,
      SendPort onExit,
      SendPort onError,
      bool errorsAreFatal,
      bool checked,
      Map<String, String> environment,
      @Deprecated('The packages/ dir is not supported in Dart 2')
          Uri packageRoot,
      Uri packageConfig,
      bool automaticPackageResolution: false,
      @Since("2.3")
          String debugName});

   //暂停/恢复
  Capability pause([Capability resumeCapability]) { ..
  external void _pause(Capability resumeCapability);
  external void resume(Capability resumeCapability);

 //拦截退出事件,获取取完消息再退出
  external void addOnExitListener(SendPort responsePort, {Object response});
 // 释放上面的钩子
  external void removeOnExitListener(SendPort responsePort);
 
//设置位置的异常是否需要终止调当前的isolate,这个调用依赖于`terminateCapability`,
//由于isolate并行运行，很有可能出错, 建议使用正确的方式来初始化isolte，设置暂停后再执行改方法,然后再开启isolate
  external void setErrorsFatal(bool errorsAreFatal);

  //是否要在事件执行完之前杀调isolate
  external void kill({int priority: beforeNextEvent});

   //请求isolate的 response数据 
// priority： immediate/ beforeNextEvent
  external void ping(SendPort responsePort,
      {Object response, int priority: immediate});
 
  external void removeErrorListener(SendPort port);
  
//从当前的isolate定义一个广播流通知
  Stream get errors {
    StreamController controller;
    RawReceivePort port;
    void handleError(message) {
      List listMessage = message;
      String errorDescription = listMessage[0];
      String stackDescription = listMessage[1];
      var error = new RemoteError(errorDescription, stackDescription);
      controller.addError(error, error.stackTrace);
    }

    controller = new StreamController.broadcast(
        sync: true,
        onListen: () {
          port = new RawReceivePort(handleError);
          this.addErrorListener(port.sendPort);
        },
        onCancel: () {
          this.removeErrorListener(port.sendPort);
          port.close();
          port = null;
        });
    return controller.stream;
  }
}

基本用法

单次任务,2个Isolate只交互处理一次,任务就完成了


void isolateTest1() async {

  ReceivePort receivePort = ReceivePort();

  //创建一个isoLater  **The entry-point function is invoked in the new isolate with [message] as the only argument.**

  await Isolate.spawn(entryPoint, receivePort.sendPort);

   //

  receivePort.listen((data) {

    print("receivePort: ${Isolate.current.hashCode}");

    if (data is SendPort) {

      data.send('receivePort hello!');

    } else {

      print('current isolate: $data');

    }

  });

}

//entryPoint是一个函数指针,在新的Ioslate创建之后首先执行它,所以它的堆栈空间肯在新创建的isolate上面

void entryPoint(SendPort sendPort) {

  ReceivePort r = ReceivePort();

  sendPort.send(r.sendPort);

  print("entryPoint: ${Isolate.current.hashCode}");

  r.listen((data) {

    print('new isolate: $data');

    sendPort.send('entryPoint hello!');

  });

}

输入出日志如下:

main isolate: 174209548
entryPoint: 1004649793

receivePort: 174209548

new isolate: receivePort hello!

receivePort: 174209548

current isolate: entryPoint hello!

持续任务,2个Isolate持续交互,如处理网络请求,音视频数据解码.

void main() async {
  await isolateTest1();

}

//创建一个Isolate进行双向通信

void isolateTest1() async {

  ReceivePort messageRecievedPort = ReceivePort();

  RawReceivePort onExitRecievedPort = RawReceivePort((exitData) {

    print('${Isolate.current.debugName} onExitRecievedPort $exitData ');

  });

  RawReceivePort onErrorRecievedPort = RawReceivePort((errorData) {

    print('${Isolate.current.debugName} onErrorRecievedPort $errorData ');

  });

  //创建一个isoLater  **The entry-point function is invoked in the new isolate with [message] as the only argument.**

  //根据flutter framework的描述,`entryPoint`

  /**

   * @param1:  `Future<Isolate>` 返回一个新的Isolate对象

   * @param2:  `void Function(T) entryPoint`, 新的Isolate的入口函数,Isolate创建完成后会首先执行此方法

   * @param3: 用于传递给新的Isolate的数据,作`void Function(T) entryPoint`的参数

   * @param4: `bool paused = false`初始化创建Isolate是否先将其暂停

   * @param5: `bool errorsAreFatal = true`,指定Isolate的错误判定行为

   * @param6: `SendPort onExit` sendport退出时的回调

   * @param7: `SendPort onError`错误时的回调

   * @param8: `Isolate`的名字

   * entryPoint 

   * Future<Isolate> spawn<T>(void Function(T) entryPoint, T message, {bool paused = false, bool errorsAreFatal = true, SendPort onExit, SendPort onError, String debugName})

  */

  await Isolate.spawn(

    entryPoint,

    messageRecievedPort.sendPort,

    onExit: onExitRecievedPort.sendPort,

    onError: onErrorRecievedPort.sendPort,

    debugName: 'malimahong',

  );

  print('1. current Isolate: ${Isolate.current.debugName} call spawn method create new Isolate');

  SendPort sendPort;

  messageRecievedPort.listen((data) async {

    if (data is SendPort) {

      sendPort = data;

      print('3. current Isolate: ${Isolate.current.debugName} recieved ????');

      print('4. current Isolate: ${Isolate.current.debugName} send agree ????');

      sendPort.send('agree ????');

    } else { 

      await Future.delayed(Duration(seconds:1));

      print('${Isolate.current.debugName} send $data -');

      sendPort.send('data -'); 

    }

  });

}

工作流程

执行流行如下:
1. current Isolate: main call spawn method create new Isolate.
2. new Isolate malimahong pass sendPort to main ,say hello, start ????
3. current Isolate: main recieved ????
4. current Isolate: main send agree ????
5. current Isolate: malimahong reieved agree ???? complete ????
  malimahong send +
  main send send + -
  malimahong send data - +
可以类比成简单的握手行为，在初始化分配新的Isolate空间后,它首先找到自己的入口函数entryPoint,入口函数包含了上一个Isolate带来的参数,如果这个参数只是简单的消息，那么只能接收消息,无法实时通信,所以这个参数必须是一个函数指针.
根据输出日志顺序可以知道在新的isolate初次创建时，首先会调用第一个参数void entryPoint(T message)，所以这个方法的实现部分函数的堆栈信息都在新建的Isolate中.除此以外的部分就是另一个isolate. 通过指定各自的receivePort然后就能快的和外成isolate通信了.