Как передавать данные между изолятами?

Как передавать данные между изолятами?

Isolates в Dart — это отдельные потоки выполнения с собственной памятью. Они не могут делиться объектами, только передавать данные через сообщения.

Основной механизм: Message Passing

Isolates обмениваются данными через SendPort и ReceivePort:

import 'dart:isolate';

void main() async {
  // Создаём ReceivePort в главном isolate
  final receivePort = ReceivePort();
  
  // Запускаем отдельный isolate, передавая его SendPort
  await Isolate.spawn(
    heavyTask,
    receivePort.sendPort,
  );
  
  // Слушаем сообщения от изолята
  receivePort.listen((message) {
    print('From isolate: $message');
  });
}

// Функция, которая выполняется в отдельном изоляте
void heavyTask(SendPort sendPort) {
  // Отправляем данные обратно в главный изолят
  sendPort.send('Task completed');
  sendPort.send({'result': 42, 'status': 'success'});
}

Двусторонняя коммуникация

Способ 1: Каналы через SendPort

import 'dart:isolate';

void main() async {
  final mainReceive = ReceivePort();
  
  await Isolate.spawn(workerTask, mainReceive.sendPort);
  
  // Получаем SendPort от изолята
  final workerSend = await mainReceive.first as SendPort;
  
  // Отправляем запрос
  final response = ReceivePort();
  workerSend.send(['calculate', 5, 10, response.sendPort]);
  
  // Получаем результат
  final result = await response.first;
  print('Result: $result');  // 15
  
  mainReceive.close();
}

void workerTask(SendPort mainSend) {
  final isolateReceive = ReceivePort();
  mainSend.send(isolateReceive.sendPort);  // Отправляем свой SendPort
  
  isolateReceive.listen((message) {
    final [operation, a, b, SendPort responseSend] = message;
    
    if (operation == 'calculate') {
      final result = a + b;
      responseSend.send(result);
    }
  });
}

Способ 2: Класс для упрощения коммуникации

class WorkerConnection {
  late SendPort _workerSend;
  final ReceivePort _mainReceive = ReceivePort();
  
  Future<void> initialize() async {
    await Isolate.spawn(_workerEntryPoint, _mainReceive.sendPort);
    _workerSend = await _mainReceive.first as SendPort;
  }
  
  Future<T> send<T>(dynamic message) async {
    final responsePort = ReceivePort();
    _workerSend.send([message, responsePort.sendPort]);
    return await responsePort.first as T;
  }
  
  static void _workerEntryPoint(SendPort mainSend) {
    final workerReceive = ReceivePort();
    mainSend.send(workerReceive.sendPort);
    
    workerReceive.listen((message) async {
      final [data, SendPort responseSend] = message;
      final result = await processData(data);
      responseSend.send(result);
    });
  }
  
  static Future<String> processData(dynamic data) async {
    return 'Processed: $data';
  }
}

// Использование:
void main() async {
  final worker = WorkerConnection();
  await worker.initialize();
  
  final result = await worker.send<String>('Hello from main');
  print(result);  // Processed: Hello from main
}

Типы данных, которые можно передавать

Поддерживаемые типы:

// Примитивы
int: 42
double: 3.14
String: "Hello"
bool: true
Null: null

// Коллекции
List: [1, 2, 3]
Map: {'key': 'value'}
Set: {1, 2, 3}

// Специальные типы
Uint8List, Int32List, Int64List, etc.
TypedData

void main() async {
  final receive = ReceivePort();
  await Isolate.spawn(workerTask, receive.sendPort);
  
  final message = await receive.first;
  print(message);  // Все эти типы могут быть переданы
}

void workerTask(SendPort send) {
  send.send(42);                          // ✓ int
  send.send('Hello');                     // ✓ String
  send.send([1, 2, 3]);                   // ✓ List
  send.send({'name': 'John', 'age': 30}); // ✓ Map
  send.send(Uint8List.fromList([1, 2]));  // ✓ TypedData
}

Поддерживаемые НЕ типы:

// ❌ Нельзя передавать
Function: () => print('Hi')           // Функции
Class instances: User()               // Объекты (в большинстве случаев)
Future: Future.value(42)              // Futures
Stream: Stream.fromIterable([1,2,3])  // Streams

// Нужно сериализовать в примитивы
void workerTask(SendPort send) {
  final user = User(name: 'John', age: 30);
  
  // ❌ Неправильно
  // send.send(user);
  
  // ✅ Правильно - конвертируем в Map
  send.send({
    'name': user.name,
    'age': user.age,
  });
}

Практический пример: обработка больших файлов

import 'dart:isolate';

class FileProcessor {
  static Future<List<String>> processLargeFile(String filePath) async {
    final receive = ReceivePort();
    
    await Isolate.spawn(
      _processFileWorker,
      {'filePath': filePath, 'sendPort': receive.sendPort},
    );
    
    // Получаем результат
    final result = await receive.first as List<String>;
    return result;
  }
  
  static void _processFileWorker(Map<String, dynamic> params) async {
    final filePath = params['filePath'] as String;
    final sendPort = params['sendPort'] as SendPort;
    
    try {
      // Тяжёлая обработка без блокировки главного потока
      final lines = await _readAndProcessFile(filePath);
      sendPort.send(lines);
    } catch (e) {
      sendPort.send({'error': e.toString()});
    }
  }
  
  static Future<List<String>> _readAndProcessFile(String path) async {
    // Симуляция тяжёлой работы
    await Future.delayed(Duration(seconds: 2));
    return ['Line 1', 'Line 2', 'Line 3'];
  }
}

// Использование:
void main() async {
  final lines = await FileProcessor.processLargeFile('large_file.txt');
  print('Processed lines: $lines');
  // UI остаётся отзывчивым во время обработки
}

Обработка ошибок

Future<T> sendWithErrorHandling<T>(dynamic message) async {
  final response = ReceivePort();
  
  try {
    _workerSend.send([message, response.sendPort]);
    
    // Таймаут защиты
    final result = await response.first
        .timeout(Duration(seconds: 30));
    
    if (result is Map && result.containsKey('error')) {
      throw Exception('Worker error: ${result['error']}');
    }
    
    return result as T;
  } on TimeoutException catch (_) {
    throw Exception('Worker request timed out');
  } finally {
    response.close();
  }
}

Лучшие практики

1. Используйте классы для управления коммуникацией:

// ✅ Хорошо - инкапсуляция
class ComputeWorker {
  late SendPort _workerPort;
  
  Future<void> initialize() async { ... }
  Future<T> compute<T>(dynamic task) async { ... }
}

2. Обрабатывайте ошибки:

// ✅ Всегда проверяйте ошибки
final result = await worker.send(data);
if (result is Map && result.containsKey('error')) {
  // Обработка ошибки
}

3. Сериализуйте сложные объекты:

// ✅ Конвертируйте в примитивы
final userData = {
  'id': user.id,
  'name': user.name,
  'email': user.email,
};
await worker.send(userData);

4. Используйте Isolate.run для простых задач:

// ✅ Для одноразовых вычислений
final result = await Isolate.run(() {
  return expensiveCalculation();
});

Message passing — это мощный способ использовать многоядерность без сложности шэринга памяти.