Dart单个ReceivePort如何实现多订阅 解决Isolate通信报错问题

问题核心原因 #

你的代码存在三个核心错误，和是否使用广播流没有关联：

• 工作Isolate仅监听单次消息：你在readAndMapFast方法中使用了receivePort.first，仅会响应第一次传入的任务，处理完成后就停止监听，后续主端发的消息无法被接收，自然会抛出异常。
• 主Isolate的控制端口监听提前终止：你主端用await for遍历控制端口流，收到所有任务结果后直接break退出了监听逻辑，后续Isolate回传的消息无法被处理，再发新任务也无法拿到返回结果。
• 未持久化存储工作Isolate的SendPort：你第一次拿到Isolate回传的自身SendPort后没有持久化存储，下次需要发新任务时找不到对应的发信端口，所以才会误以为要靠广播流解决问题。

修复方案 #

步骤1：定义持久化存储变量

在State类中定义全局复用的端口、Isolate发信端口列表等变量：

class _MainScreenState extends State<MainScreen> {
  final ImagePicker _picker = ImagePicker();
  Future<ui.Image>? image;
  // 主控制端口，全局唯一
  final ReceivePort controlPort = ReceivePort();
  // 存储所有工作Isolate的发信端口
  final List<SendPort> workerSendPorts = [];
  // 设备处理器核心数
  late final int numberOfWorkers = Platform.numberOfProcessors;
  // 结果计数、结果存储变量
  int counter = 0;
  List<FinalList> postEditList = [];
  BytesBuilder bytesBuilder = BytesBuilder();

步骤2：修改工作Isolate逻辑为持续监听

不要用first单次监听，改成循环监听消息，支持多次任务处理，同时新增关闭指令支持资源释放：

static void readAndMapFast(SendPort mainSendPort) async {
  final ReceivePort isolateReceivePort = ReceivePort();
  // 将当前Isolate的发信端口回传给主端
  mainSendPort.send(isolateReceivePort.sendPort);

  // 持续监听主端发来的消息
  await for (final message in isolateReceivePort) {
    if (message is Arguments) {
      // 执行任务逻辑
      final output = yourTaskProcessLogic(message);
      // 回传结果
      mainSendPort.send(output);
    } else if (message == 'close') {
      // 收到关闭指令时释放资源
      isolateReceivePort.close();
      break;
    }
  }
}

步骤3：主端仅初始化一次端口监听

不要每次发任务都重新开启监听，initState中只做一次监听，全程复用：

@override
void initState() {
  super.initState();
  // 全局唯一的控制端口监听，不要break不要重复初始化
  controlPort.listen((message) {
    if (message is SendPort) {
      // 收集工作Isolate的发信端口
      workerSendPorts.add(message);
      return;
    }
    if (message is FinalList) {
      counter++;
      postEditList.add(message);
      // 收齐所有worker的结果后处理逻辑
      if (counter == numberOfWorkers) {
        final Stopwatch timerr = Stopwatch()..start();
        for (int t = 0; t < postEditList.length; t++) {
          final FinalList list = postEditList.firstWhere((element) => element.order == t);
          bytesBuilder.add(list.imgSubList);
        }
        timerr.stop();
        // 处理完成后重置计数和结果列表，为下一次任务做准备
        counter = 0;
        postEditList.clear();
      }
    }
  });
  // 启动所有工作Isolate
  for (int i = 0; i < numberOfWorkers; i++) {
    Isolate.spawn(Core.readAndMapFast, controlPort.sendPort);
  }
}

步骤4：发任务直接复用已存储的SendPort

需要下发新任务时，直接从提前存储的workerSendPorts里拿端口发消息即可，不需要再做端口相关的初始化操作：

void dispatchNewTask(List<Arguments> argsList) {
  // 保证所有Isolate已经完成初始化
  assert(workerSendPorts.length == numberOfWorkers);
  for (int i = 0; i < numberOfWorkers; i++) {
    workerSendPorts[i].send(argsList[i]);
  }
}

步骤5：页面销毁时释放资源

避免内存泄漏：

@override
void dispose() {
  // 通知所有工作Isolate关闭
  for (final sendPort in workerSendPorts) {
    sendPort.send('close');
  }
  // 关闭主控制端口
  controlPort.close();
  super.dispose();
}

内容的提问来源于stack exchange，提问作者Dinmukhamed Isenov