불안정한 행을 각각 따로 재시도하기

FxDart 승 async

요구사항

불안정한 엔드포인트를 통해 행 여섯 개를 임포트합니다: 짝수 행은 성공하기 전에 정확히 한 번 실패합니다. 각 행에 자기만의 재시도 예산 두 번을 주고, 한 번에 최대 세 행씩 실행하며, 결과를 소스 순서대로 성공한 시도 횟수와 함께 출력하세요. 실패 주입과 행별 지연은 결정적이며 코드에 들어 있습니다; 두 버전 모두 예상 출력 아래에 표시된 줄들을 출력해야 합니다.

예상 출력

row 1 (alpha) imported on attempt 1
row 2 (bravo) imported on attempt 2
row 3 (charlie) imported on attempt 1
row 4 (delta) imported on attempt 2
row 5 (echo) imported on attempt 1
row 6 (foxtrot) imported on attempt 2

나란히 보기

RxDart

import 'package:rxdart/rxdart.dart';

/// Six import rows; the even rows hit a flaky endpoint that fails exactly
/// once before succeeding. Per-row delays make completion order differ
/// from source order under concurrency.
const rows = [
  (1, 'alpha'),
  (2, 'bravo'),
  (3, 'charlie'),
  (4, 'delta'),
  (5, 'echo'),
  (6, 'foxtrot'),
];
const importMs = [0, 70, 30, 50, 30, 20, 40]; // indexed by row id

final attemptsByRow = <int, int>{};

Future<String> importRow((int, String) row) async {
  final (id, name) = row;
  final attempt = attemptsByRow.update(id, (n) => n + 1, ifAbsent: () => 1);
  await Future<void>.delayed(Duration(milliseconds: importMs[id]));
  if (id.isEven && attempt == 1) throw StateError('endpoint reset on $name');
  return 'row $id ($name) imported on attempt $attempt';
}

Future<void> main() async {
  final results = await Stream.fromIterable(rows)
      // Each row gets its own retrying inner stream, 3 subscribed at a
      // time — but flatMap emits in COMPLETION order, so every result
      // must carry its row id for the sort back to source order.
      .flatMap(
          (row) => Rx.retry(() => Rx.fromCallable(() => importRow(row)), 1)
              .map((line) => (row.$1, line)),
          maxConcurrent: 3)
      .toList();

results.sort((a, b) => a.$1.compareTo(b.$1));
  for (final (_, line) in results) {
    print(line);
  }
}

FxDart

import 'package:fxdart/fxdart.dart';

final attemptsByRow = <int, int>{};

Future<void> main() async {
  // Two attempts per row, three rows in flight — each in-flight row
  // retries independently, and mapRetry under concurrent yields in
  // SOURCE order by construction.
  final results = await fx(rows)
      .toAsync()
      .mapRetry(2, importRow)
      .concurrent(3)
      .toList();

results.forEach(print);
}

차이가 나는 이유

회복탄력성 절반은 양쪽이 같은 방식으로 표현합니다: 행마다 재시도 래퍼를 씌워, 불안정한 행 하나가 다시 도는 동안 이웃 행들은 그대로 지나갑니다. RxDart는 이를 maxConcurrent: 3과 함께 행마다 재시도하는 내부 스트림으로의 flatMap으로 쓰고, FxDart는 concurrent(3) 아래의 mapRetry(2, …)로 씁니다 — 진행 중인 각 요소가 자기만의 독립적인 예산을 지니는 형태입니다.

차이는 반대쪽 끝에서 무엇이 나오는가입니다. flatMap은 내부 스트림들을 완료 순서로 병합합니다 — 그것이 이 연산자의 계약이므로 — 세 행이 동시에 진행되고 지연이 제각각이면 결과는 뒤섞여 도착합니다. 소스 순서를 되찾으려면 모든 결과에 행 id를 태그로 붙이고 toList 뒤에 정렬해야 합니다. 순서를 지켜 주는 RxDart 연산자인 concatMap은 동시성을 포기하는 대가로 그렇게 합니다 — 한 번에 한 행씩. 풀 모델에서는 순서 있는 동시성이 기본 모드입니다: concurrent(3)은 세 개의 풀을 한꺼번에 평가하되 구조적으로 소스 순서대로 내놓으므로, 태그 붙일 것도 정렬할 것도 없습니다.

판정은 FxDart — 순서가 이야기의 핵심입니다. "N개씩, 각각 독립적으로 재시도, 순서대로"는 풀 모델에서는 체인 하나이고, 푸시 모델에서는 병합한 뒤 다시 줄 세우는 우회책입니다.

벤치마크

Apple M1 Max, RAM 32 GB · Dart 3.12.2 (AOT 컴파일) · 2026-08-18

비동기 예제입니다. 대표 스케일이 1,000,000이 아니라 N = 10,000인 이유 — 원소 하나마다 양쪽 모두 이벤트 루프를 한 바퀴 돌아야 하므로, 실제 await 백만 번은 파이프라인이 아니라 Dart 이벤트 루프를 몇 분씩 재게 됩니다. 지연은 0으로 두고 예제의 동시성 제한은 그대로 유지합니다. 막대가 비교하는 것은 파이프라인 기계 장치 자체입니다.

N = 100

시간 무승부

RxDart 737 µs

FxDart 768 µs

최대 메모리 무승부

RxDart 17.2 MB

FxDart 16.7 MB

N = 10,000

시간 무승부

RxDart 65.4 ms

FxDart 62.9 ms

최대 메모리 FxDart 승

RxDart 61.2 MB

FxDart 40.3 MB

막대는 사이드별로 새 프로세스에서 반복 측정한 중앙값입니다(작은 N은 타이머 해상도를 위해 배치 처리). 두 사이드가 서로 5% 이내이거나 — 사람이 지각할 수 없는 차이인 0.6ms 이내이면 — 무승부로 칩니다. 상대 차이가 근소한 경우는 최대 5회까지 다시 측정합니다. 앱에서는 어느 막대가 짧든 몇 밀리초 이하의 차이는 사용자에게 보이지 않습니다. 메모리는 프로세스 최대 RSS입니다. Dart VM과 데이터셋은 양쪽이 동일하므로, 두 막대의 차이가 곧 파이프라인 자체가 붙들고 있는 양입니다.