Isolates

Isolates

Dart에서 isolate를 작성하는 방법을 살펴볼게요. 계산이 커서 다른 계산을 잠시 블로킹할 때 isolate가 왜 유용한지, 실제 API를 예시와 함께 풀어봐요.

출처: Dart 공식 문서

본문

이 페이지는 Isolate API를 사용해 isolate를 구현하는 몇 가지 예시를 다뤄요.

애플리케이션이 다른 계산을 임시로 블로킹할 만큼 큰 계산을 처리할 때는 isolate를 사용해야 해요. 가장 흔한 예시는 Flutter 애플리케이션에서, UI를 응답하지 않게 만들 수 있는 큰 계산을 수행해야 할 때예요.

Flutter 안내

Flutter 웹에서는 여러 isolate가 지원되지 않아요. 참고: 웹에서의 동시성

isolate를 반드시 사용해야 하는 규칙 같은 건 없지만, 유용할 수 있는 상황이 몇 가지 더 있어요.

• 비정상적으로 큰 JSON blob을 파싱하고 디코딩하기.
• 사진, 오디오, 비디오를 처리하고 압축하기.
• 오디오·비디오 파일 변환하기.
• 큰 리스트나 파일 시스템에서 복잡한 검색·필터링 수행하기.
• 데이터베이스와 통신 같은 I/O 수행하기.
• 대량의 네트워크 요청 처리하기.

간단한 워커 isolate 구현하기 (Implementing a simple worker isolate)

이 예시들은 간단한 워커 isolate를 스폰하는 main isolate를 구현해요. Isolate.run()이 워커 isolate를 설정하고 관리하는 단계를 단순화해 줘요.

• isolate를 스폰(시작하고 생성)합니다.
• 스폰된 isolate에서 함수를 실행합니다.
• 결과를 포착합니다.
• 결과를 main isolate로 반환합니다.
• 작업이 끝나면 isolate를 종료합니다.
• 예외와 오류를 검사·포착하고 main isolate로 다시 던집니다.

Flutter 안내

Flutter를 사용한다면 Isolate.run() 대신 Flutter의 compute 함수를 사용할 수 있어요.

새 isolate에서 기존 메서드 실행하기 (Running an existing method in a new isolate)

• main()이 결과를 기다리는 동안 main isolate에서 직접 run()을 호출해 새 isolate(백그라운드 워커)를 스폰해요.

const String filename = 'with_keys.json';

void main() async {
  // Read some data.
  final jsonData = await Isolate.run(_readAndParseJson);

  // Use that data.
  print('Number of JSON keys: ${jsonData.length}');
}

• 워커 isolate에 실행하길 원하는 함수를 첫 번째 인자로 전달해요. 이 예시에서는 기존 함수 _readAndParseJson()을 전달해요.

Future<Map<String, dynamic>> _readAndParseJson() async {
  final fileData = await File(filename).readAsString();
  final jsonData = jsonDecode(fileData) as Map<String, dynamic>;
  return jsonData;
}

• Isolate.run()은 _readAndParseJson()이 반환한 결과를 받아 그 값을 main isolate로 보내고, 워커 isolate를 종료해요.
• 워커 isolate는 결과를 담고 있는 메모리를 main isolate로 전송(transfer) 해요. 데이터를 복사하지 않아요. 워커 isolate는 객체가 전송되도록 허용되는지 확인하기 위해 검증 단계를 수행해요.

_readAndParseJson()은 기존의 비동기 함수로, main isolate에서 직접 실행해도 그만이에요. 그것을 Isolate.run()으로 실행하면 동시성이 가능해져요. 워커 isolate는 _readAndParseJson()의 계산을 완전히 추상화해요. main isolate를 블로킹하지 않고 완료될 수 있어요.

Isolate.run()의 결과는 항상 Future예요. main isolate의 코드는 계속 실행되기 때문이죠. 워커 isolate가 실행하는 계산이 동기든 비동기든 main isolate에는 영향을 주지 않아요. 어느 쪽이든 동시에 실행되니까요.

전체 프로그램은 send_and_receive.dart 샘플을 확인해 보세요.

isolate로 클로저 보내기 (Sending closures with isolates)

또한 run()으로 main isolate에서 바로 함수 리터럴, 즉 클로저를 사용해 간단한 워커 isolate를 만들 수도 있어요.

const String filename = 'with_keys.json';

void main() async {
  // Read some data.
  final jsonData = await Isolate.run(() async {
    final fileData = await File(filename).readAsString();
    final jsonData = jsonDecode(fileData) as Map<String, dynamic>;
    return jsonData;
  });

  // Use that data.
  print('Number of JSON keys: ${jsonData.length}');
}

이 예시는 앞선 예시와 같은 일을 해요. 새 isolate가 스폰되고, 무언가를 계산하고, 결과를 보내요.

다만 이제 isolate는 클로저를 보내요. 클로저는 동작 방식과 코드에서 작성되는 방식 모두에서 일반적인 이름 있는 함수보다 제약이 덜해요. 이 예시에서 Isolate.run()은 로컬 코드처럼 보이는 것을 동시에 실행해요. 그런 의미에서 run()이 '병렬 실행'을 위한 제어 흐름 연산자처럼 동작한다고 상상할 수 있어요.

포트로 isolate 사이에 여러 메시지 보내기 (Sending multiple messages between isolates with ports)

수명이 짧은 isolate는 사용하기 편리하지만, 새 isolate를 스폰하고 한 isolate에서 다른 isolate로 객체를 복사하는 데 성능 오버헤드가 필요해요. 코드가 Isolate.run으로 같은 계산을 반복해서 실행한다면, 즉시 종료되지 않는 오래 살아있는(long-lived) isolate를 만드는 편이 성능을 개선할 수 있어요.

이를 위해 Isolate.run이 추상화한 저수준 isolate API 몇 가지를 사용할 수 있어요.

• Isolate.spawn()와 Isolate.exit()
• ReceivePort와 SendPort
• SendPort.send() 메서드

이 절은 새로 스폰한 isolate와 main isolate 사이에 양방향(2-way) 통신을 구축하는 데 필요한 단계를 다뤄요. 첫 번째 예시인 Basic ports가 그 과정을 높은 수준에서 소개해요. 두 번째 예시인 Robust ports는 첫 번째 예시에 점차 더 실용적이고 실제 세계적인 기능을 더해요.

ReceivePort와 SendPort

isolate 사이의 오래 살아있는 통신을 설정하려면(Isolate 외에) ReceivePort와 SendPort라는 두 클래스가 필요해요. 이 포트들이 isolate가 서로 소통할 수 있는 유일한 방법이에요.

ReceivePort는 다른 isolate에서 보낸 메시지를 처리하는 객체예요. 그 메시지들은 SendPort를 통해 보내져요.

정보

SendPort 객체는 정확히 하나의 ReceivePort와 연결되지만, 하나의 ReceivePort는 많은 SendPort를 가질 수 있어요. ReceivePort를 만들면 자신을 위한 SendPort가 생성돼요. 기존 ReceivePort에 메시지를 보낼 수 있는 추가 SendPort를 만들 수도 있어요.

포트는 Stream 객체와 비슷하게 동작해요. (실제로 receive port는 Stream을 구현해요!) SendPort와 ReceivePort를 각각 Stream의 StreamController와 리스너라고 생각하면 돼요. SendPort는 SendPort.send() 메서드로 메시지를 '추가'하므로 StreamController와 같고, 그 메시지들은 리스너, 이 경우 ReceivePort가 처리해요. ReceivePort는 받은 메시지를 당신이 제공한 콜백에 인자로 전달해 처리해요.

포트 설정하기 (Setting up ports)

새로 스폰된 isolate는 Isolate.spawn 호출을 통해 받은 정보만 가져요. main isolate가 스폰된 isolate와 초기 생성 이후에도 계속 소통해야 한다면, 스폰된 isolate가 main isolate로 메시지를 보낼 수 있는 통신 채널을 설정해야 해요. isolate는 메시지 전달로만 소통할 수 있어요. 서로의 메모리 '안을 볼 수' 없기 때문이고, 그래서 이름이 'isolate(격리)'인 거예요.

이 양방향 통신을 설정하려면 먼저 main isolate에 ReceivePort를 만들고, Isolate.spawn으로 새 isolate를 스폰할 때 그 SendPort를 인자로 전달해요. 그러면 새 isolate는 자신의 ReceivePort를 만들고, main isolate가 전달한 SendPort를 통해 자신의 SendPort를 돌려보내요. main isolate는 이 SendPort를 받고, 이제 양쪽 모두 메시지를 보내고 받을 열린 채널을 가지게 돼요.

정보

이 절의 다이어그램은 높은 수준이고 isolate에 포트를 사용하는 개념을 전달하기 위한 거예요. 실제 구현은 이 페이지 뒷부분에 나오는 것처럼 코드가 조금 더 필요해요.

• main isolate에서 ReceivePort를 만들어요. SendPort는 ReceivePort의 속성으로 자동 생성돼요.
• Isolate.spawn()으로 워커 isolate를 스폰해요.
• ReceivePort.sendPort에 대한 참조를 워커 isolate에 첫 번째 메시지로 전달해요.
• 워커 isolate에서 또 다른 새 ReceivePort를 만들어요.
• 워커 isolate의 ReceivePort.sendPort에 대한 참조를 main isolate에 첫 번째 메시지로 돌려보내요.

포트를 만들고 통신을 설정하는 것과 함께, 포트가 메시지를 받을 때 무엇을 할지도 알려줘야 해요. 이는 각 ReceivePort의 listen 메서드로 수행해요.

• 워커 isolate의 SendPort에 대한 main isolate의 참조를 통해 메시지를 보내요.
• 워커 isolate의 ReceivePort의 리스너를 통해 메시지를 받고 처리해요. 이 지점에서 main isolate에서 옮기고 싶은 계산이 실행돼요.
• main isolate의 SendPort에 대한 워커 isolate의 참조를 통해 반환 메시지를 보내요.
• main isolate의 ReceivePort의 리스너를 통해 메시지를 받아요.

Basic ports 예시 (Basic ports example)

이 예시는 main isolate와 그 사이에 양방향 통신이 있는 오래 살아있는 워커 isolate를 어떻게 설정하는지 보여줘요. 코드는 JSON 텍스트를 새 isolate로 보내고, 그곳에서 JSON이 파싱·디코딩된 후 main isolate로 돌아오는 예시를 사용해요.

경고

이 예시는 시간이 지나며 여러 메시지를 보내고 받을 수 있는 새 isolate를 스폰하는 데 필요한 아주 최소한을 가르치기 위한 거예요.

오류 처리, 포트 종료, 메시지 순서 지정 같은 프로덕션 소프트웨어에서 기대되는 중요한 기능들은 다루지 않아요.

다음 절의 Robust ports 예시가 이러한 기능과, 그것이 없을 때 생길 수 있는 몇몇 문제를 다뤄요.

1단계: 워커 클래스 정의하기 (Step 1: Define the worker class)

먼저 백그라운드 워커 isolate용 클래스를 만들어요. 이 클래스에는 필요한 모든 기능이 들어 있는데, 다음을 수행해요.

• isolate를 스폰한다.
• 그 isolate에 메시지를 보낸다.
• isolate가 일부 JSON을 디코딩하게 한다.
• 디코딩된 JSON을 main isolate로 보낸다.

이 클래스는 두 개의 public 메서드를 노출해요. 하나는 워커 isolate를 스폰하고, 다른 하나는 그 워커 isolate에 메시지를 보내는 걸 처리해요.

이 예시의 나머지 절에서 클래스 메서드를 하나씩 채우는 방법을 보여줄게요.

class Worker {
  Future<void> spawn() async {
    // TODO: Add functionality to spawn a worker isolate.
  }

  void _handleResponsesFromIsolate(dynamic message) {
    // TODO: Handle messages sent back from the worker isolate.
  }

  static void _startRemoteIsolate(SendPort port) {
    // TODO: Define code that should be executed on the worker isolate.
  }

  Future<void> parseJson(String message) async {
    // TODO: Define a public method that can
    // be used to send messages to the worker isolate.
  }
}

2단계: 워커 isolate 스폰하기 (Step 2: Spawn a worker isolate)

Worker.spawn 메서드는 워커 isolate를 만들고 그것이 메시지를 받고 보낼 수 있게 하는 코드를 모으는 곳이에요.

• 먼저 ReceivePort를 만들어요. 이렇게 하면 main isolate가 새로 스폰된 워커 isolate에서 보낸 메시지를 받을 수 있어요.
• 다음으로 리시브 포트에 리스너를 추가해 워커 isolate가 돌려보낼 메시지를 처리해요. 리스너에 전달된 콜백 _handleResponsesFromIsolate는 4단계에서 다뤄요.
• 마지막으로 Isolate.spawn으로 워커 isolate를 스폰해요. 이는 두 개의 인자를 기대해요: 워커 isolate에서 실행될 함수(3단계에서 다룸)와 리시브 포트의 sendPort 속성.

Future<void> spawn() async {
  final receivePort = ReceivePort();
  receivePort.listen(_handleResponsesFromIsolate);
  await Isolate.spawn(_startRemoteIsolate, receivePort.sendPort);
}

receivePort.sendPort 인자는 워커 isolate에서 콜백(_startRemoteIsolate)이 호출될 때 그 콜백에 인자로 전달돼요. 이는 워커 isolate가 main isolate로 메시지를 보낼 방법을 가지도록 보장하는 첫 단계예요.

3단계: 워커 isolate에서 코드 실행하기 (Step 3: Execute code on the worker isolate)

이 단계에서는 워커 isolate가 스폰될 때 실행되도록 그 isolate에 보내지는 메서드 _startRemoteIsolate를 정의해요. 이 메서드는 워커 isolate의 'main' 메서드와 같아요.

• 먼저 또 다른 새 ReceivePort를 만들어요. 이 포트는 main isolate로부터의 향후 메시지를 받아요.
• 다음으로 그 포트의 SendPort를 main isolate로 보내요.
• 마지막으로 새 ReceivePort에 리스너를 추가해요. 이 리스너는 main isolate가 워커 isolate로 보내는 메시지를 처리해요.

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

  receivePort.listen((dynamic message) async {
    if (message is String) {
      final transformed = jsonDecode(message);
      port.send(transformed);
    }
  });
}

워커의 ReceivePort에 있는 리스너는 main isolate에서 전달된 JSON을 디코딩한 다음, 디코딩된 JSON을 main isolate로 보내요.

이 리스너는 main isolate에서 워커 isolate로 보내는 메시지의 진입점이에요. 이것이 워커 isolate에 앞으로 실행할 코드를 알려줄 유일한 기회예요.

4단계: main isolate에서 메시지 처리하기 (Step 4: Handle messages on the main isolate)

마지막으로, 워커 isolate가 main isolate로 보낸 메시지를 처리하는 방법을 main isolate에 알려줘야 해요. 이를 위해 _handleResponsesFromIsolate 메서드를 채워야 해요. 이 메서드는 2단계에서 설명한 대로 receivePort.listen 메서드에 전달된다는 점을 기억하세요.

Future<void> spawn() async {
  final receivePort = ReceivePort();
  receivePort.listen(_handleResponsesFromIsolate);
  await Isolate.spawn(_startRemoteIsolate, receivePort.sendPort);
}

또한 3단계에서 SendPort를 main isolate로 보냈다는 것도 기억하세요. 이 메서드는 그 SendPort의 수신뿐 아니라 향후 메시지(디코딩된 JSON이 될) 처리도 담당해요.

• 먼저 메시지가 SendPort인지 확인해요. 그렇다면 그 포트를 클래스의 _sendPort 속성에 할당해 나중에 메시지를 보내는 데 사용할 수 있게 해요.
• 다음으로 메시지가 Map<String, dynamic> 타입인지 확인해요. 디코딩된 JSON의 예상 타입이에요. 그렇다면 애플리케이션 특정 로직으로 그 메시지를 처리해요. 이 예시에서는 메시지를 출력해요.

void _handleResponsesFromIsolate(dynamic message) {
  if (message is SendPort) {
    _sendPort = message;
    _isolateReady.complete();
  } else if (message is Map<String, dynamic>) {
    print(message);
  }
}

5단계: isolate가 준비됐는지 확인하는 Completer 추가하기 (Step 5: Add a completer to ensure your isolate is set-up)

클래스를 완성하려면 워커 isolate에 메시지를 보내는 책임을 가진 parseJson이라는 public 메서드를 정의해요. 그것은 또한 isolate가 완전히 준비되기 전에 메시지를 보낼 수 있는지도 보장해야 해요. 이를 처리하려면 Completer를 사용해요.

• 먼저 Completer라는 클래스 수준 속성을 추가하고 _isolateReady라고 이름 지어요.
• 다음으로 (4단계에서 만든) _handleResponsesFromIsolate 메서드에, 메시지가 SendPort라면 completer의 complete() 호출을 추가해요.
• 마지막으로 parseJson 메서드에서 _sendPort.send 앞에 await _isolateReady.future를 추가해요. 이렇게 하면 워커 isolate가 스폰되고 자신의 SendPort를 main isolate로 보낼 때까지 어떤 메시지도 워커 isolate로 보낼 수 없음이 보장돼요.

Future<void> parseJson(String message) async {
  await _isolateReady.future;
  _sendPort.send(message);
}

완성된 예시 (Complete example)

전체 예시를 볼게요.

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

void main() async {
  final worker = Worker();
  await worker.spawn();
  await worker.parseJson('{"key":"value"}');
}

class Worker {
  late SendPort _sendPort;
  final Completer<void> _isolateReady = Completer.sync();

  Future<void> spawn() async {
    final receivePort = ReceivePort();
    receivePort.listen(_handleResponsesFromIsolate);
    await Isolate.spawn(_startRemoteIsolate, receivePort.sendPort);
  }

  void _handleResponsesFromIsolate(dynamic message) {
    if (message is SendPort) {
      _sendPort = message;
      _isolateReady.complete();
    } else if (message is Map<String, dynamic>) {
      print(message);
    }
  }

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

    receivePort.listen((dynamic message) async {
      if (message is String) {
        final transformed = jsonDecode(message);
        port.send(transformed);
      }
    });
  }

  Future<void> parseJson(String message) async {
    await _isolateReady.future;
    _sendPort.send(message);
  }

}

Robust ports 예시 (Robust ports example)

이전 예시는 양방향 통신이 있는 오래 살아있는 isolate를 설정하는 데 필요한 기본 구성 요소를 설명했어요. 언급했듯이 그 예시에는 오류 처리, 더 이상 사용하지 않을 때 포트를 닫는 기능, 일부 상황에서 메시지 순서의 불일치 같은 중요한 기능이 빠져 있어요.

이 예시는 더 나은 설계 패턴을 따르면서, 앞서 말한 기능들 이상을 가진 오래 살아있는 워커 isolate를 만들어 첫 번째 예시의 정보를 확장해요. 이 코드는 첫 번째 예시와 유사점이 있지만, 그 예시의 확장은 아니에요.

정보

이 예시는 이전 예시에서 다룬 Isolate.spawn과 포트로 isolate 사이의 통신 구축에 이미 익숙하다고 가정해요.

1단계: 워커 클래스 정의하기 (Step 1: Define the worker class)

먼저 백그라운드 워커 isolate용 클래스를 만들어요. 이 클래스에는 필요한 모든 기능이 들어 있는데, 다음을 수행해요.

• isolate를 스폰한다.
• 그 isolate에 메시지를 보낸다.
• isolate가 일부 JSON을 디코딩하게 한다.
• 디코딩된 JSON을 main isolate로 보낸다.

이 클래스는 세 개의 public 메서드를 노출해요. 하나는 워커 isolate를 만들고, 하나는 그 워커 isolate에 메시지를 보내는 걸 처리하며, 하나는 더 이상 사용하지 않을 때 포트를 종료할 수 있어요.

class Worker {
  final SendPort _commands;
  final ReceivePort _responses;

  Future<Object?> parseJson(String message) async {
    // TODO: Ensure the port is still open.
    _commands.send(message);
  }

  static Future<Worker> spawn() async {
    // TODO: Add functionality to create a new Worker object with a
    //  connection to a spawned isolate.
    throw UnimplementedError();
  }

  Worker._(this._responses, this._commands) {
    // TODO: Initialize main isolate receive port listener.
  }

  void _handleResponsesFromIsolate(dynamic message) {
    // TODO: Handle messages sent back from the worker isolate.
  }

  static void _handleCommandsToIsolate(ReceivePort rp, SendPort sp) async {
    // TODO: Handle messages sent back from the worker isolate.
  }

  static void _startRemoteIsolate(SendPort sp) {
    // TODO: Initialize worker isolate's ports.
  }
}

정보

이 예시에서 SendPort와 ReceivePort 인스턴스는 main isolate를 기준으로 이름을 짓는 모범 사례 명명 규칙을 따르고 있어요. main isolate에서 워커 isolate로 SendPort를 통해 보내는 메시지를 commands(명령) 라고 하고, main isolate로 되돌아오는 메시지를 responses(응답) 라고 불러요.

2단계: Worker.spawn 메서드에 RawReceivePort 만들기 (Step 2: Create a RawReceivePort in the Worker.spawn method)

isolate를 스폰하기 전에 RawReceivePort를 만들어야 해요. 이것은 더 저수준의 ReceivePort예요. RawReceivePort를 사용하는 것이 선호되는 패턴인데, isolate 시작 로직과 isolate에서 메시지 전달을 처리하는 로직을 분리할 수 있게 해주기 때문이에요.

Worker.spawn 메서드에서:

• 먼저 RawReceivePort를 만들어요. 이 ReceivePort는 워커 isolate로부터의 초기 메시지(그것은 SendPort가 될)를 받는 것만 담당해요.
• 다음으로 isolate가 메시지를 받을 준비가 됐음을 알리는 Completer를 만들어요. 이것이 완료되면 ReceivePort와 SendPort가 든 record를 반환해요.
• 다음으로 RawReceivePort.handler 속성을 정의해요. 이 속성은 ReceivePort.listener처럼 동작하는 Function?이에요. 이 포트가 메시지를 받으면 이 함수가 호출돼요.
• 핸들러 함수 안에서 connection.complete()을 호출해요. 이 메서드는 인자로 ReceivePort와 SendPort가 든 record를 기대해요. SendPort는 워커 isolate에서 보낸 초기 메시지이며, 다음 단계에서 클래스 수준의 _commands라는 SendPort에 할당돼요.
• 그런 다음 ReceivePort.fromRawReceivePort 생성자로 새 ReceivePort를 만들고 initPort를 전달해요.

class Worker {
  final SendPort _commands;
  final ReceivePort _responses;

  static Future<Worker> spawn() async {
    // Create a receive port and add its initial message handler.
    final initPort = RawReceivePort();
    final connection = Completer<(ReceivePort, SendPort)>.sync();
    initPort.handler = (initialMessage) {
      final commandPort = initialMessage as SendPort;
      connection.complete((
        ReceivePort.fromRawReceivePort(initPort),
        commandPort,
      ));
    };
  }
}

RawReceivePort를 먼저 만들고 나서 ReceivePort를 만들면, 나중에 ReceivePort.listen에 새 콜백을 추가할 수 있게 돼요. 반대로 ReceivePort를 곧바로 만들면 ReceivePort가 Stream(BroadcastStream이 아니라)을 구현하므로 리스너를 하나만 추가할 수 있어요.

효과적으로 이렇게 하면 isolate 시작 로직과 통신 설정이 끝난 뒤 메시지 수신을 처리하는 로직을 분리할 수 있어요. 이 이점은 다른 메서드의 로직이 커질수록 더 분명해져요.

3단계: Isolate.spawn으로 워커 isolate 스폰하기 (Step 3: Spawn a worker isolate with Isolate.spawn)

이 단계는 Worker.spawn 메서드를 계속 채워요. isolate를 스폰하고 이 클래스에서 Worker 인스턴스를 반환하는 데 필요한 코드를 추가해요. 이 예시에서 Isolate.spawn 호출은 try/catch 블록으로 감싸고, isolate가 시작에 실패하면 initPort가 닫히고 Worker 객체가 만들어지지 않도록 해요.

• 먼저 try/catch 블록에서 워커 isolate를 스폰하려 시도해요. 워커 isolate 스폰이 실패하면 이전 단계에서 만든 리시브 포트를 닫아요. Isolate.spawn에 전달된 메서드는 이후 단계에서 다뤄요.
• 다음으로 connection.future를 await하고, 그것이 반환하는 record에서 send port와 receive port를 구조 분해해요.
• 마지막으로 비공개 생성자를 호출하고 그 completer의 포트들을 전달해 Worker 인스턴스를 반환해요.

class Worker {
  final SendPort _commands;
  final ReceivePort _responses;

  static Future<Worker> spawn() async {
    // Create a receive port and add its initial message handler.
    final initPort = RawReceivePort();
    final connection = Completer<(ReceivePort, SendPort)>.sync();
    initPort.handler = (initialMessage) {
      final commandPort = initialMessage as SendPort;
      connection.complete((
        ReceivePort.fromRawReceivePort(initPort),
        commandPort,
      ));
    };
    // Spawn the isolate.
    try {
      await Isolate.spawn(_startRemoteIsolate, (initPort.sendPort));
    } on Object {
      initPort.close();
      rethrow;
    }

    final (ReceivePort receivePort, SendPort sendPort) =
        await connection.future;

    return Worker._(receivePort, sendPort);
  }
}

이전 예시와 비교해 이 예시에서 Worker.spawn은 이 클래스의 비동기 정적 생성자 역할을 하며 Worker 인스턴스를 만드는 유일한 방법이에요. 이렇게 하면 API가 단순해져 Worker 인스턴스를 만드는 코드가 더 깔끔해져요.

4단계: isolate 설정 과정 완성하기 (Step 4: Complete the isolate setup process)

이 단계에서는 기본 isolate 설정 과정을 완성해요. 이는 이전 예시와 거의 완전히 대응하고 새 개념은 없어요. 코드가 더 많은 메서드로 쪼개졌다는 약간의 변화가 있는데, 이는 이 예시의 나머지 부분을 통해 더 많은 기능을 추가할 준비를 해주는 설계 관행이에요. isolate를 설정하는 기본 과정에 대한 심층 설명은 basic ports 예시를 보세요.

먼저 Worker.spawn 메서드에서 반환되는 비공개 생성자를 만들어요. 생성자 본문에서 main isolate가 사용하는 리시브 포트에 리스너를 추가하고, 아직 정의되지 않은 _handleResponsesFromIsolate라는 메서드를 그 리스너에 전달해요.

class Worker {
  final SendPort _commands;
  final ReceivePort _responses;

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

다음으로 워커 isolate의 포트 초기화를 담당하는 코드를 _startRemoteIsolate에 추가해요. 이 메서드는 Worker.spawn 메서드에서 Isolate.spawn에 전달됐고, main isolate의 SendPort를 인자로 받게 된다는 점을 기억하세요.

• 새 ReceivePort를 만들어요.
• 그 포트의 SendPort를 main isolate로 보내요.
• _handleCommandsToIsolate라는 새 메서드를 호출하고 새 ReceivePort와 main isolate의 SendPort를 인자로 전달해요.

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

다음으로 main isolate에서 메시지를 받고, 워커 isolate에서 json을 디코딩하며, 디코딩된 json을 응답으로 보내는 _handleCommandsToIsolate 메서드를 추가해요.

• 먼저 워커 isolate의 ReceivePort에 리스너를 선언해요.
• 리스너에 추가된 콜백 안에서, main isolate에서 전달된 JSON을 try/catch 블록 안에서 디코딩하려 시도해요. 디코딩에 성공하면 디코딩된 JSON을 main isolate로 보내요.
• 오류가 있으면 RemoteError를 돌려보내요.

static void _handleCommandsToIsolate(
  ReceivePort receivePort,
  SendPort sendPort,
) {
  receivePort.listen((message) {
    try {
      final jsonData = jsonDecode(message as String);
      sendPort.send(jsonData);
    } catch (e) {
      sendPort.send(RemoteError(e.toString(), ''));
    }
  });
}

다음으로 _handleResponsesFromIsolate 메서드의 코드를 추가해요.

• 먼저 메시지가 RemoteError인지 확인하고, 그렇다면 그 오류를 던져요.
• 그렇지 않으면 메시지를 출력해요. 이후 단계에서 이 코드를 메시지를 출력 대신 반환하도록 업데이트할 거예요.

void _handleResponsesFromIsolate(dynamic message) {
  if (message is RemoteError) {
    throw message;
  } else {
    print(message);
  }
}

마지막으로 바깥 코드가 JSON을 워커 isolate로 보내 디코딩하게 하는 public 메서드인 parseJson을 추가해요.

Future<Object?> parseJson(String message) async {
  _commands.send(message);
}

이 메서드는 다음 단계에서 업데이트할 거예요.

5단계: 여러 메시지를 동시에 처리하기 (Step 5: Handle multiple messages at the same time)

현재 워커 isolate에 메시지를 빠르게 보내면, isolate는 보낸 순서가 아니라 완료되는 순서대로 디코딩된 json 응답을 보내요. 어떤 응답이 어떤 메시지에 해당하는지 알 방법이 없어요.

이 단계에서는 각 메시지에 id를 부여하고 Completer 객체를 사용해, 바깥 코드가 parseJson을 호출할 때 그 호출자에게 반환되는 응답이 올바른 응답임을 보장하는 방식으로 이 문제를 고쳐요.

먼저 Worker에 두 개의 클래스 수준 속성을 추가해요.

• Map<int, Completer<Object?>> _activeRequests
• int _idCounter

class Worker {
  final SendPort _commands;
  final ReceivePort _responses;
  final Map<int, Completer<Object?>> _activeRequests = {};
  int _idCounter = 0;
  // ···
}

_activeRequests 맵은 워커 isolate로 보낸 메시지를 Completer와 연결해요. _activeRequests에 쓰이는 키는 _idCounter에서 가져오며, 메시지를 더 보낼수록 증가해요.

다음으로 parseJson 메서드를 업데이트해 워커 isolate에 메시지를 보내기 전에 completer를 만들게 해요.

• 먼저 Completer를 만들어요.
• 다음으로 _idCounter를 증가시켜 각 Completer가 고유한 숫자와 연결되게 해요.
• _activeRequests 맵에 키가 현재 _idCounter 값이고 값이 completer인 항목을 추가해요.
• id와 함께 메시지를 워커 isolate로 보내요. SendPort로는 하나의 값만 보낼 수 있으므로, id와 메시지를 record로 감싸요.
• 마지막으로 completer의 future를 반환해요. 이것은 결국 워커 isolate의 응답을 담게 될 거예요.

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

또한 이 시스템을 처리하도록 _handleResponsesFromIsolate와 _handleCommandsToIsolate를 업데이트해야 해요.

_handleCommandsToIsolate에서 message가 json 텍스트만이 아니라 두 값을 가진 record가 된 것을 반영해야 해요. message에서 값을 구조 분해해요.

그다음 json을 디코딩한 후 sendPort.send 호출을 업데이트해 id와 디코딩된 json을 다시 record를 사용해 main isolate로 전달해요.

static void _handleCommandsToIsolate(
  ReceivePort receivePort,
  SendPort sendPort,
) {
  receivePort.listen((message) {
    final (int id, String jsonText) = message as (int, String); // New
    try {
      final jsonData = jsonDecode(jsonText);
      sendPort.send((id, jsonData)); // Updated
    } catch (e) {
      sendPort.send((id, RemoteError(e.toString(), '')));
    }
  });
}

마지막으로 _handleResponsesFromIsolate를 업데이트해요.

• 먼저 message 인자에서 id와 응답을 다시 구조 분해해요.
• 그런 다음 _activeRequests 맵에서 이 요청에 해당하는 completer를 제거해요.
• 마지막으로 오류를 던지거나 디코딩된 json을 출력하는 대신, 응답을 전달하며 completer를 완성해요. 이게 완료되면 응답이 main isolate에서 parseJson을 호출한 코드로 반환돼요.

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

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

6단계: 포트를 닫는 기능 추가하기 (Step 6: Add functionality to close the ports)

더 이상 코드가 isolate를 사용하지 않을 때는 main isolate와 워커 isolate의 포트를 닫아야 해요.

• 먼저 포트가 닫혔는지 추적하는 클래스 수준 불리언을 추가해요.
• 그런 다음 Worker.close 메서드를 추가해요. 이 메서드 안에서:
  • _closed를 true로 업데이트해요.
  • 워커 isolate에 마지막 메시지를 보내요. 이 메시지는 "shutdown"이라는 String이지만, 원하는 어떤 객체든 될 수 있어요. 다음 코드 조각에서 사용할 거예요.
• 마지막으로 _activeRequests가 비어 있는지 확인해요. 비어 있다면 main isolate의 _responses라는 ReceivePort를 닫아요.

class Worker {
  bool _closed = false;
  // ···
  void close() {
    if (!_closed) {
      _closed = true;
      _commands.send('shutdown');
      if (_activeRequests.isEmpty) _responses.close();
      print('--- port closed --- ');
    }
  }
}

• 다음으로 워커 isolate에서 "shutdown" 메시지를 처리해야 해요. _handleCommandsToIsolate 메서드에 다음 코드를 추가해요. 이 코드는 메시지가 "shutdown"이라는 String인지 확인해요. 그렇다면 워커 isolate의 ReceivePort를 닫고 반환해요.

static void _handleCommandsToIsolate(
  ReceivePort receivePort,
  SendPort sendPort,
) {
  receivePort.listen((message) {
    // New if-block.
    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(), '')));
    }
  });
}

• 마지막으로 메시지를 보내기 전에 포트가 닫혔는지 확인하는 코드를 추가해야 해요. Worker.parseJson 메서드에 한 줄을 추가해요.

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

완성된 예시 (Complete example)

전체 예시를 볼게요.

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

void main() async {
  final worker = await Worker.spawn();
  print(await worker.parseJson('{"key":"value"}'));
  print(await worker.parseJson('"banana"'));
  print(await worker.parseJson('[true, false, null, 1, "string"]'));
  print(
    await Future.wait([worker.parseJson('"yes"'), worker.parseJson('"no"')]),
  );
  worker.close();
}

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

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

  static Future<Worker> spawn() async {
    // Create a receive port and add its initial message handler.
    final initPort = RawReceivePort();
    final connection = Completer<(ReceivePort, SendPort)>.sync();
    initPort.handler = (initialMessage) {
      final commandPort = initialMessage as SendPort;
      connection.complete((
        ReceivePort.fromRawReceivePort(initPort),
        commandPort,
      ));
    };

    // Spawn the isolate.
    try {
      await Isolate.spawn(_startRemoteIsolate, (initPort.sendPort));
    } on Object {
      initPort.close();
      rethrow;
    }

    final (ReceivePort receivePort, SendPort sendPort) =
        await connection.future;

    return Worker._(receivePort, sendPort);
  }

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

  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 _handleCommandsToIsolate(
    ReceivePort receivePort,
    SendPort 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(), '')));
      }
    });
  }

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

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

더 알아보기

• Concurrency in Dart — 이벤트 루프와 isolate 개념
• dart:isolate API 문서 — Isolate.run, Isolate.spawn, ReceivePort, SendPort
• Flutter isolates — Flutter에서의 isolate 사용