Dart: Dart Isolates — 无共享内存的并行计算

Isolate 是 Dart 的并行之道 — 无共享内存，天然无竞态，安全且高效。

1. 你将学到

Isolate vs Thread：无共享内存的并发模型
Isolate.spawn 与 ReceivePort / SendPort 双向通信
Isolate.run：简化的一次性任务 API
Compute 便利函数与 Flutter 的 compute
Bob 场景：DataPipeline 用 Isolate 并行处理百万级订单

2. 一个开发者的真实故事

(1) 痛点：单线程处理百万数据太慢

Bob 的 DataPipeline 需要对 1,200,000 条订单执行统计分析。单线程处理需要 15 秒，但 SaaS 客户要求报表在 5 秒内生成。Bob 尝试用多线程，但共享内存的锁和竞态条件让他修了一周的并发 bug，最终放弃，报表依然要 15 秒。

(2) Isolate 的解法

Dart 的 Isolate 是无共享内存的并行单元，每个 Isolate 有独立堆内存，通过消息传递通信。没有共享内存就没有竞态条件。

DART

// Split 1.2M orders into 4 Isolates, each processes 300K
final results = await Future.wait([
  Isolate.run(() => processChunk(orders.sublist(0, 300000))),
  Isolate.run(() => processChunk(orders.sublist(300000, 600000))),
  Isolate.run(() => processChunk(orders.sublist(600000, 900000))),
  Isolate.run(() => processChunk(orders.sublist(900000, 1200000))),
]);

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

(3) 收益

4 个 Isolate 并行，处理时间从 15 秒降到 4 秒
无共享内存 = 无竞态条件 = 无锁 = 无并发 bug
消息传递模型让代码更容易推理

3. Isolate 基础

(1) Isolate vs Thread

flowchart TD
  A[Main Isolate] -->|"SendPort"| B[Worker Isolate 1]
  A -->|"SendPort"| C[Worker Isolate 2]
  A -->|"SendPort"| D[Worker Isolate N]
  B -->|"SendPort"| A
  C -->|"SendPort"| A
  D -->|"SendPort"| A
  subgraph Data Sharding
    B -- 300K orders
    C -- 300K orders
    D -- 400K orders
  end

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

维度	Thread (Java/C++)	Isolate (Dart)
内存	共享	独立
通信	共享变量+锁	消息传递
竞态条件	有	无
数据同步	手动加锁	无需锁
创建开销	低	中（需拷贝数据）

4. Isolate.run — 简化的一次性任务

(1) 最简单的 Isolate 用法

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Isolate.run 基础

DART

import 'dart:isolate';

// Expensive computation to run in isolate
int fibonacci(int n) {
  if (n <= 1) return n;
  return fibonacci(n - 1) + fibonacci(n - 2);
}

Future``<void>`` main() async {
  print('Computing fibonacci(40) in isolate...');

  final result = await Isolate.run(() => fibonacci(40));

  print('Result: $result');
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：Isolate.run 处理数据块

DART

import 'dart:isolate';

// Process a chunk of orders
double processChunk(List<Map<String, dynamic>> orders) {
  double total = 0;
  for (final order in orders) {
    total += (order['amount'] as double);
  }
  return total;
}

Future``<void>`` main() async {
  // Simulate 1M orders
  final orders = List.generate(
    1000000,
    (i) => {'id': 'ORD-$i', 'amount': (i % 100 + 1) * 10.0},
  );

  // Split into 4 chunks
  final chunkSize = orders.length ~/ 4;
  final chunks = List.generate(
    4,
    (i) => orders.sublist(i * chunkSize, (i + 1) * chunkSize),
  );

  // Process chunks in parallel
  final stopwatch = Stopwatch()..start();
  final results = await Future.wait(
    chunks.map((chunk) => Isolate.run(() => processChunk(chunk))),
  );
  stopwatch.stop();

  final totalRevenue = results.fold(0.0, (a, b) => a + b);
  print('Total revenue: \$${totalRevenue.toStringAsFixed(2)} USD');
  print('Time: ${stopwatch.elapsedMilliseconds}ms');
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

5. Isolate.spawn 与双向通信

(1) SendPort / ReceivePort 通信

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：双向通信

DART

import 'dart:isolate';

// Worker function - runs in separate isolate
void workerIsolate(SendPort mainSendPort) {
  final workerReceivePort = ReceivePort();

  // Send our receive port to main isolate
  mainSendPort.send(workerReceivePort.sendPort);

  // Listen for messages from main isolate
  workerReceivePort.listen((message) {
    if (message is String && message == 'shutdown') {
      workerReceivePort.close();
      return;
    }

    // Process data and send result back
    if (message is List``<double>``) {
      final total = message.fold(0.0, (a, b) => a + b);
      mainSendPort.send(total);
    }
  });
}

Future``<void>`` main() async {
  final mainReceivePort = ReceivePort();

  // Spawn worker isolate
  await Isolate.spawn(workerIsolate, mainReceivePort.sendPort);

  // Get worker's send port
  final workerSendPort = await mainReceivePort.first as SendPort;

  // Create a new receive port for response
  final responsePort = ReceivePort();
  workerSendPort.send([1500.0, 3200.0, 890.0]);

  // Wait for response
  final result = await responsePort.first;
  print('Result from worker: $result');

  // Shutdown worker
  workerSendPort.send('shutdown');
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：多轮通信

DART

import 'dart:isolate';

void dataWorker(SendPort mainPort) {
  final receivePort = ReceivePort();
  mainPort.send(receivePort.sendPort);

  receivePort.listen((message) {
    if (message == 'done') {
      receivePort.close();
      return;
    }

    if (message is Map<String, dynamic>) {
      // Process order data
      final amount = message['amount'] as double;
      final taxRate = message['taxRate'] as double? ?? 0.08;
      final total = amount * (1 + taxRate);
      mainPort.send({'id': message['id'], 'total': total});
    }
  });
}

Future``<void>`` main() async {
  final mainPort = ReceivePort();
  await Isolate.spawn(dataWorker, mainPort.sendPort);

  final workerPort = await mainPort.first as SendPort;

  // Send multiple messages
  final responsePort = ReceivePort();
  workerPort.add({'id': 'ORD-001', 'amount': 1500.0, 'taxRate': 0.08});
  workerPort.add({'id': 'ORD-002', 'amount': 3200.0});

  // ... simplified communication
  workerPort.send('done');
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

6. Isolate 与数据传递

(1) 数据传递规则

传递方式	说明	性能
基本类型	int/double/String/bool	快速拷贝
List/Map	深度拷贝	中等
自定义对象	深度拷贝	中等
SendPort	引用传递	快速
函数闭包	Isolate.run 传递	编译时检查

⚠️ 注意： Isolate 间传递的数据会被拷贝，不是共享。大列表传递会有拷贝开销。对于大数据，考虑分块传递或使用 Isolate.run 的闭包传递。

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：传递大数据的最佳实践

DART

import 'dart:isolate';

// Pass data via closure (Isolate.run)
Future``<double>`` processLargeData(List``<double>`` amounts) async {
  return Isolate.run(() {
    // amounts is copied into the isolate
    double sum = 0;
    for (final a in amounts) {
      sum += a;
    }
    return sum;
  });
}

// Better: pass only what's needed
Future``<double>`` processChunkOptimized(List``<double>`` chunk) async {
  return Isolate.run(() => chunk.fold(0.0, (a, b) => a + b));
}

void main() async {
  final data = List.generate(1000000, (i) => (i + 1) * 1.0);

  // Split and process in parallel
  final chunkSize = data.length ~/ 4;
  final futures = List.generate(4, (i) {
    final chunk = data.sublist(i * chunkSize, (i + 1) * chunkSize);
    return Isolate.run(() => chunk.fold(0.0, (a, b) => a + b));
  });

  final results = await Future.wait(futures);
  final total = results.fold(0.0, (a, b) => a + b);
  print('Total: $total');
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

7. Bob 场景：百万订单并行处理

▶ 示例

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

：DataPipeline 并行统计

DART

import 'dart:isolate';

class ChunkResult {
  final int processed;
  final int skipped;
  final double revenue;
  final Map<String, double> categoryRevenue;

  ChunkResult({
    required this.processed,
    required this.skipped,
    required this.revenue,
    required this.categoryRevenue,
  });
}

ChunkResult processChunk(List<Map<String, dynamic>> chunk) {
  int processed = 0;
  int skipped = 0;
  double revenue = 0;
  final categoryRevenue = <String, double>{};

  for (final order in chunk) {
    final amount = order['amount'] as double;
    final status = order['status'] as String;
    final category = order['category'] as String;

    if (amount <= 0 || status == 'cancelled') {
      skipped++;
      continue;
    }

    processed++;
    revenue += amount;
    categoryRevenue.update(category, (v) => v + amount, ifAbsent: () => amount);
  }

  return ChunkResult(
    processed: processed,
    skipped: skipped,
    revenue: revenue,
    categoryRevenue: categoryRevenue,
  );
}

Future``<void>`` main() async {
  // Generate 1.2M orders
  final categories = ['Electronics', 'Books', 'Clothing', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final orders = List.generate(1200000, (i) => {
    'id': 'ORD-${i.toString().padLeft(6, '0')}',
    'amount': (i % 500 + 10) * 1.0,
    'status': statuses[i % statuses.length],
    'category': categories[i % categories.length],
  });

  // Split into chunks
  final isolateCount = 4;
  final chunkSize = orders.length ~/ isolateCount;
  final chunks = List.generate(isolateCount, (i) {
    final start = i * chunkSize;
    final end = i == isolateCount - 1 ? orders.length : (i + 1) * chunkSize;
    return orders.sublist(start, end);
  });

  // Process in parallel
  final stopwatch = Stopwatch()..start();
  final results = await Future.wait(
    chunks.map((chunk) => Isolate.run(() => processChunk(chunk))),
  );
  stopwatch.stop();

  // Aggregate results
  int totalProcessed = 0;
  int totalSkipped = 0;
  double totalRevenue = 0;
  final totalCategoryRevenue = <String, double>{};

  for (final r in results) {
    totalProcessed += r.processed;
    totalSkipped += r.skipped;
    totalRevenue += r.revenue;
    for (final entry in r.categoryRevenue.entries) {
      totalCategoryRevenue.update(entry.key, (v) => v + entry.value, ifAbsent: () => entry.value);
    }
  }

  print('=== DataPipeline Parallel Report ===');
  print('Orders:    ${orders.length}');
  print('Processed: $totalProcessed');
  print('Skipped:   $totalSkipped');
  print('Revenue:   \$${totalRevenue.toStringAsFixed(2)} USD');
  print('Time:      ${stopwatch.elapsedMilliseconds}ms');
  print('Isolates:  $isolateCount');

  print('\nBy Category:');
  for (final entry in totalCategoryRevenue.entries.toList()..sort((a, b) => b.value.compareTo(a.value))) {
    print('  ${entry.key}: \$${entry.value.toStringAsFixed(2)} USD');
  }
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

8. 完整示例：DataPipeline Isolate 并行处理框架

DART

// ============================================
// DataPipeline Isolate Parallel Processing
// Framework for parallel data analytics
// ============================================

import 'dart:isolate';

// Worker result
class WorkerResult {
  final int workerId;
  final int processed;
  final int skipped;
  final double revenue;
  final Map<String, double> categoryRevenue;
  final Duration processingTime;

  WorkerResult({
    required this.workerId,
    required this.processed,
    required this.skipped,
    required this.revenue,
    required this.categoryRevenue,
    required this.processingTime,
  });
}

// Worker function - runs in separate isolate
WorkerResult processInIsolate((int, List<Map<String, dynamic>>) input) {
  final (workerId, orders) = input;
  final stopwatch = Stopwatch()..start();

  int processed = 0;
  int skipped = 0;
  double revenue = 0;
  final categoryRevenue = <String, double>{};

  for (final order in orders) {
    final amount = (order['amount'] as num).toDouble();
    final status = order['status'] as String;
    final category = order['category'] as String;

    if (amount <= 0 || status == 'cancelled') {
      skipped++;
      continue;
    }

    processed++;
    revenue += amount;
    categoryRevenue.update(category, (v) => v + amount, ifAbsent: () => amount);
  }

  stopwatch.stop();

  return WorkerResult(
    workerId: workerId,
    processed: processed,
    skipped: skipped,
    revenue: revenue,
    categoryRevenue: categoryRevenue,
    processingTime: stopwatch.elapsed,
  );
}

// Parallel pipeline manager
class ParallelPipeline {
  final int isolateCount;

  ParallelPipeline({this.isolateCount = 4});

  Future``<void>`` process(List<Map<String, dynamic>> orders) async {
    print('=== DataPipeline Parallel Processing ===');
    print('Orders: ${orders.length}, Isolates: $isolateCount');

    // Split orders into chunks
    final chunkSize = orders.length ~/ isolateCount;
    final chunks = List.generate(isolateCount, (i) {
      final start = i * chunkSize;
      final end = i == isolateCount - 1 ? orders.length : (i + 1) * chunkSize;
      return (i, orders.sublist(start, end));
    });

    // Process in parallel
    final totalStopwatch = Stopwatch()..start();
    final results = await Future.wait(
      chunks.map((chunk) => Isolate.run(() => processInIsolate(chunk))),
    );
    totalStopwatch.stop();

    // Aggregate
    int totalProcessed = 0;
    int totalSkipped = 0;
    double totalRevenue = 0;
    final totalCategoryRevenue = <String, double>{};

    print('\nWorker Results:');
    for (final r in results) {
      totalProcessed += r.processed;
      totalSkipped += r.skipped;
      totalRevenue += r.revenue;
      for (final entry in r.categoryRevenue.entries) {
        totalCategoryRevenue.update(
          entry.key, (v) => v + entry.value, ifAbsent: () => entry.value);
      }
      print('  Worker ${r.workerId}: ${r.processed} processed, '
          '${r.processingTime.inMilliseconds}ms');
    }

    print('\n--- Aggregate Report ---');
    print('Processed:  $totalProcessed orders');
    print('Skipped:    $totalSkipped records');
    print('Revenue:    \$${totalRevenue.toStringAsFixed(2)} USD');
    print('Wall time:  ${totalStopwatch.elapsedMilliseconds}ms');

    print('\nBy Category:');
    final sorted = totalCategoryRevenue.entries.toList()
      ..sort((a, b) => b.value.compareTo(a.value));
    for (final entry in sorted) {
      final pct = (entry.value / totalRevenue * 100).toStringAsFixed(1);
      print('  ${entry.key}: \$${entry.value.toStringAsFixed(2)} USD ($pct%)');
    }
  }
}

void main() async {
  // Generate sample data
  final categories = ['Electronics', 'Books', 'Clothing', 'Home', 'Sports'];
  final statuses = ['completed', 'completed', 'completed', 'pending', 'cancelled'];
  final orders = List.generate(500000, (i) => <String, dynamic>{
    'id': 'ORD-${i.toString().padLeft(6, '0')}',
    'amount': (i % 500 + 10) * 1.0,
    'status': statuses[i % statuses.length],
    'category': categories[i % categories.length],
  });

  final pipeline = ParallelPipeline(isolateCount: 4);
  await pipeline.process(orders);
}

TEXT 📖 仅展示

> **输出：** 在本地 DartPad 或 `dart run` 执行。Dart 课程所有示例基于 Dart 3.x / Flutter 3.x，运行结果会因 SDK 版本略有差异。

❓ 常见问题

Q：Isolate 和线程有什么区别？ A：Isolate 有独立的堆内存，不共享数据；Thread 共享堆内存。Isolate 通过消息传递通信，Thread 通过共享变量+锁通信。Isolate 没有竞态条件。

Q：Isolate 的创建开销大吗？ A：创建一个 Isolate 大约需要 50-150ms，比创建线程慢。频繁创建销毁不经济，推荐使用 Isolate Pool 或 Isolate.run（自动管理生命周期）。

Q：Isolate.run 和 Isolate.spawn 有什么区别？ A：Isolate.run 是简化 API，执行一次性任务后自动关闭 Isolate；Isolate.spawn 创建持久 Isolate，需要手动管理生命周期和通信。

Q：数据在 Isolate 间传递会被拷贝吗？ A：是的。所有传递的数据都会被深度拷贝（除了 SendPort）。大列表传递有性能开销。对于超大数据，考虑分块传递。

Q：Dart 有多少种 Isolate？ A：主要两种：Isolate（通用并行）和 Flutter 的 compute（UI 隔离）。Web 平台不支持真正的 Isolate，使用 Web Worker 模拟。

Q：Isolate 数量有没有上限？ A：没有硬性上限，但每个 Isolate 占用约 2MB 内存。实际建议不超过 CPU 核心数，避免过多上下文切换。

Q：Isolate 适合什么场景？ A：CPU 密集型计算（数据分析、图像处理、加密计算）。I/O 密集型用 Future/Stream 即可，不需要 Isolate。

📖 小节

Isolate 是 Dart 的并行单元，独立内存，消息传递通信，天然无竞态
Isolate.run 适合一次性计算任务，自动管理生命周期
Isolate.spawn + SendPort/ReceivePort 适合持久通信
数据传递会被拷贝，大数据需分块传递
4 个 Isolate 并行处理百万订单，时间从 15 秒降到 4 秒

📝 作业

基础题（难度⭐）：用 Isolate.run 计算斐波那契数列的第 42 项，与主线程同步计算对比时间。
进阶题（难度⭐⭐）：生成 1,000,000 个随机数，分成 4 份，用 4 个 Isolate 并行计算每份的总和和平均值，最后在主 Isolate 汇总。
挑战题（难度⭐⭐⭐）：用 Isolate.spawn 实现一个 Isolate Pool：预先创建 N 个 Worker Isolate，主 Isolate 通过 SendPort 分发任务，Worker 处理完返回结果。支持任务队列和负载均衡。

← 上一课 | 下一课 →

Web-Tutorial 技术团队

由多位开发者共同维护的编程教程平台。每篇教程由对应领域的开发者编写和审核，确保内容准确可靠。如发现任何问题，欢迎向我们反馈。

关于我们

Dart: Dart Isolates — 无共享内存的并行计算

1. 你将学到

2. 一个开发者的真实故事

(1) 痛点：单线程处理百万数据太慢

(2) Isolate 的解法

(3) 收益

3. Isolate 基础

(1) Isolate vs Thread

4. Isolate.run — 简化的一次性任务

(1) 最简单的 Isolate 用法

▶ 示例

▶ 示例

5. Isolate.spawn 与双向通信

(1) SendPort / ReceivePort 通信

▶ 示例

▶ 示例

6. Isolate 与数据传递

(1) 数据传递规则

▶ 示例

7. Bob 场景：百万订单并行处理

▶ 示例

8. 完整示例：DataPipeline Isolate 并行处理框架

❓ 常见问题

📖 小节

📝 作业

🙏 帮我们做得更好