dart-concurrency-isolates

管理 Dart 并发与 Isolate

目录

• 核心准则
• 选择合适的 Isolate 策略
• 实现一次性任务
• 实现长时间运行的 Worker
• 工作流
• 示例

核心准则

• **Isolate 内存：**假设 Isolate 之间完全不共享内存。Isolate 只能通过消息传递进行通信。
• **数据传输：**避免在 Isolate 之间传递大型可变对象。优先使用简单数据类型或不可变记录，以尽量减少序列化开销。
• **资源管理：**不再需要 Isolate 和端口时，务必将其终止，以防止内存泄漏。
• **平台限制：**不要在 Dart Web 平台上使用 Isolate。Web 会编译为 JavaScript，而 JavaScript 使用 Web Workers。
• **相关技能：**有关在主 Isolate 上运行的标准异步操作（Future、Stream、async/await），请参阅 dart-async-programming。

选择合适的 Isolate 策略

根据条件逻辑确定合适的 Isolate 实现方式：

• 如果要执行简单的一次性后台任务（例如，解析单个大型 JSON 负载、压缩文件）：使用 Isolate.run()。
• 如果要执行复杂的长时间运行后台 Worker（例如，持续处理数据、维护持久数据库连接）：使用 Isolate.spawn()，并手动管理 SendPort 和 ReceivePort。

实现一次性任务

使用 Isolate.run() 创建 Isolate、执行函数、获取结果，并自动终止 Isolate。

1. 将顶层函数、静态方法或闭包传递给 Isolate.run()。
2. 在主 Isolate 中等待结果。

Future<Map<String, dynamic>> parseLargeJson(String jsonString) async {
  // Spawns isolate, runs decode, returns result, and terminates automatically.
  return await Isolate.run(() => jsonDecode(jsonString) as Map<String, dynamic>);
}

实现长时间运行的 Worker

手动管理 SendPort 和 ReceivePort，为长时间运行的 Worker Isolate 建立双向通信。

1. **使用 RawReceivePort 初始化：**在主 Isolate 中使用 RawReceivePort，将启动逻辑与持续的消息处理分开。
2. **建立双向通信：**通过 Isolate.spawn() 将主 Isolate 的 SendPort 传递给 Worker Isolate。Worker Isolate 必须创建自己的 ReceivePort，并将自己的 SendPort 发回主 Isolate。
3. **将请求映射到响应：**使用带有唯一 ID 的 Completer 映射，跟踪发送给 Worker Isolate 的异步请求，并在收到响应时完成这些请求。
4. **处理错误：**捕获 Worker Isolate 中的异常，并将其作为 RemoteError 对象发回。
5. **清理：**向 Worker Isolate 发送特定的关闭消息，以关闭其 ReceivePort；所有活动请求完成后，关闭主 Isolate 的 ReceivePort。

工作流

任务进度：设置长时间运行的 Worker

实现长时间运行的 Worker Isolate 时，复制此清单以跟踪进度：

• 创建 Worker 类以封装 Isolate 管理。
• 实现静态 spawn() 方法，使用 RawReceivePort 接收 Worker 发来的初始 SendPort。
• 调用 Isolate.spawn()，传入 RawReceivePort.sendPort 和 Worker 入口方法。
• 实现 Worker 入口方法（_startRemoteIsolate）。
• 在 Worker 中创建 ReceivePort，并将其 SendPort 发回主 Isolate。
• 在 Worker 的 ReceivePort 上设置监听器，以处理传入的命令。
• 在主 Isolate 的 ReceivePort 上设置监听器，以处理响应并完成 Completer 实例。
• 实现 close() 方法以发送关闭命令并关闭所有端口。
• 运行验证器 -> 检查错误 -> 修复（确保没有内存泄漏，且所有端口都能正常关闭）。

示例

健壮的长时间运行 Worker 实现

使用此模式与长时间运行的 Worker Isolate 进行可靠的双向通信。

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

class JsonWorker {
  final SendPort _commands;
  final ReceivePort _responses;
  final Map<int, Completer<Object?>> _activeRequests = {};
  int _idCounter = 0;
  bool _closed = false;

  JsonWorker._(this._responses, this._commands) {
    _responses.listen(_handleResponsesFromIsolate);
  }

  static Future<JsonWorker> spawn() async {
    final initPort = RawReceivePort();
    final connection = Completer<(ReceivePort, SendPort)>.sync();
    
    initPort.handler = (initialMessage) {
      final commandPort = initialMessage as SendPort;
      connection.complete((
        ReceivePort.fromRawReceivePort(initPort),
        commandPort,
      ));
    };

    try {
      await Isolate.spawn(_startRemoteIsolate, initPort.sendPort);
    } catch (e) {
      initPort.close();
      rethrow;
    }

    final (ReceivePort receivePort, SendPort sendPort) = await connection.future;
    return JsonWorker._(receivePort, sendPort);
  }

  Future<Object?> parseJson(String message) async {
    if (_closed) throw StateError('Worker is closed');
    final completer = Completer<Object?>.sync();
    final id = _idCounter++;
    _activeRequests[id] = completer;
    _commands.send((id, message));
    return await completer.future;
  }

  void _handleResponsesFromIsolate(dynamic message) {
    final (int id, Object? response) = message as (int, Object?);
    final completer = _activeRequests.remove(id)!;

    if (response is RemoteError) {
      completer.completeError(response);
    } else {
      completer.complete(response);
    }

    if (_closed && _activeRequests.isEmpty) _responses.close();
  }

  static void _startRemoteIsolate(SendPort sendPort) {
    final receivePort = ReceivePort();
    sendPort.send(receivePort.sendPort);

    receivePort.listen((message) {
      if (message == 'shutdown') {
        receivePort.close();
        return;
      }
      
      final (int id, String jsonText) = message as (int, String);
      try {
        final jsonData = jsonDecode(jsonText);
        sendPort.send((id, jsonData));
      } catch (e) {
        sendPort.send((id, RemoteError(e.toString(), '')));
      }
    });
  }

  void close() {
    if (!_closed) {
      _closed = true;
      _commands.send('shutdown');
      if (_activeRequests.isEmpty) _responses.close();
    }
  }
}