Dart: Isolates في Dart — الحوسبة المتوازية بدون ذاكرة مشتركة

// تقسيم 1.2 مليون طلب إلى 4 Isolates، كل منها يعالج 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))),
]);

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

import 'dart:isolate';

// حساب مكلف للتشغيل في 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');
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

import 'dart:isolate';

// معالجة جزء من الطلبات
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 {
  // محاكاة مليون طلب
  final orders = List.generate(
    1000000,
    (i) => {'id': 'ORD-$i', 'amount': (i % 100 + 1) * 10.0},
  );

  // التقسيم إلى 4 أجزاء
  final chunkSize = orders.length ~/ 4;
  final chunks = List.generate(
    4,
    (i) => orders.sublist(i * chunkSize, (i + 1) * chunkSize),
  );

  // معالجة الأجزاء بالتوازي
  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');
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

import 'dart:isolate';

// دالة العامل - تعمل في isolate منفصل
void workerIsolate(SendPort mainSendPort) {
  final workerReceivePort = ReceivePort();

  // أرسل منفذ الاستقبال الخاص بنا إلى main isolate
  mainSendPort.send(workerReceivePort.sendPort);

  // الاستماع للرسائل من main isolate
  workerReceivePort.listen((message) {
    if (message is String && message == 'shutdown') {
      workerReceivePort.close();
      return;
    }

    // معالجة البيانات وإرسال النتيجة مرة أخرى
    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();

  // إنشاء worker isolate
  await Isolate.spawn(workerIsolate, mainReceivePort.sendPort);

  // الحصول على send port للعامل
  final workerSendPort = await mainReceivePort.first as SendPort;

  // إنشاء منفذ استقبال جديد للاستجابة
  final responsePort = ReceivePort();
  workerSendPort.send([1500.0, 3200.0, 890.0]);

  // انتظار الاستجابة
  final result = await responsePort.first;
  print('Result from worker: $result');

  // إيقاف العامل
  workerSendPort.send('shutdown');
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

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>) {
      // معالجة بيانات الطلب
      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;

  // إرسال رسائل متعددة
  final responsePort = ReceivePort();
  workerPort.add({'id': 'ORD-001', 'amount': 1500.0, 'taxRate': 0.08});
  workerPort.add({'id': 'ORD-002', 'amount': 3200.0});

  // ... تواصل مبسط
  workerPort.send('done');
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

import 'dart:isolate';

// تمرير البيانات عبر الإغلاق (Isolate.run)
Future<double> processLargeData(List<double> amounts) async {
  return Isolate.run(() {
    // amounts يتم نسخه إلى الـ isolate
    double sum = 0;
    for (final a in amounts) {
      sum += a;
    }
    return sum;
  });
}

// أفضل: مرر فقط ما هو مطلوب
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);

  // التقسيم والمعالجة بالتوازي
  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');
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

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 {
  // توليد 1.2 مليون طلب
  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],
  });

  // التقسيم إلى أجزاء
  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);
  });

  // المعالجة بالتوازي
  final stopwatch = Stopwatch()..start();
  final results = await Future.wait(
    chunks.map((chunk) => Isolate.run(() => processChunk(chunk))),
  );
  stopwatch.stop();

  // تجميع النتائج
  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');
  }
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.

// ============================================
// المعالجة المتوازية بـ Isolate لخط أنابيب البيانات
// إطار لتحليلات البيانات المتوازية
// ============================================

import 'dart:isolate';

// نتيجة العامل
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,
  });
}

// دالة العامل - تعمل في 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,
  );
}

// مدير خط أنابيب متوازي
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');

    // تقسيم الطلبات إلى أجزاء
    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));
    });

    // المعالجة بالتوازي
    final totalStopwatch = Stopwatch()..start();
    final results = await Future.wait(
      chunks.map((chunk) => Isolate.run(() => processInIsolate(chunk))),
    );
    totalStopwatch.stop();

    // التجميع
    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 {
  // توليد بيانات نموذجية
  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);
}

> **الإخراج:** شغّل في DartPad محلي أو عبر `dart run`. جميع أمثلة Dart مبنية على Dart 3.x / Flutter 3.x. قد تختلف النتائج قليلاً مع إصدارات SDK المختلفة.