동시성 처리

concurrent는 여러 비동기 값을 한 번에 처리할 수 있는 함수입니다.

JavaScript에는 Promise.all로 여러 promise 값을 동시에 평가하는 함수가 있습니다. 그러나 동시 요청의 부하를 처리할 수 없고 무한한 열거 가능한 데이터 집합에 대한 요청을 처리할 수 없습니다. concurrent는 무한 데이터 집합의 비동기 요청을 처리하고 부하의 요청 크기를 제어할 수 있습니다.

// prettier-ignore
import { pipe, toAsync, range, map, filter, take, each, concurrent } from "@fxts/core";

const fetchApi = (page) =>
  new Promise((resolve) => setTimeout(() => resolve(page), 1000));

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi), // 0,1,2,3,4,5
  filter((a) => a % 2 === 0),
  take(3), // 0,2,4
  concurrent(3), // 이 줄이 없으면 총 6초가 걸립니다.
  each(console.log), // 2초
);

하나씩 요청할 때 6초가 걸리지만 concurrent를 사용하면 2초가 걸리는 것을 확인할 수 있습니다.

유용한 예제

더 실용적인 코드는 아래와 같습니다.

위 코드에서 concurrent의 위치가 아래와 같다면, 결과가 달라질까요? 아닙니다, 동일합니다! concurrent는 항상 길이가 변경되기 전의 Iterable에 적용된다는 점에 유의하세요.

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi),
  concurrent(3),
  filter((a) => a % 2 === 0),
  take(3),
  each(console.log),
);

map까지 순차적으로 하나씩 평가하고, filter의 비동기 predicate를 세 개씩 동시에 평가하고 싶다면, 아래 코드를 작성해야 합니다:

await pipe(
  range(Infinity),
  toAsync,
  map(fetchApi),
  toArray,
  filter((a) => delay(100, a % 2 === 0)),
  take(3),
  concurrent(3),
  each(console.log),
);

윈도우 vs 풀: concurrent와 concurrentPool

concurrent(n)은 고정 윈도우 단위로 평가합니다: n개의 요청을 발사하고, 윈도우가 끝나기를 기다린 뒤 다음 윈도우를 시작합니다 — 비슷한 크기의 배치형 작업에 이상적입니다.

작업 시간이 제각각이면 윈도우는 가장 느린 멤버를 기다리게 됩니다. concurrentPool(n)은 대신 롤링 풀을 유지합니다 — 하나가 끝나는 즉시 다음이 시작되며, 결과는 여전히 입력 순서대로 내보냅니다:

await pipe(
  toAsync(urls),
  map(fetchItem),
  concurrentPool(5), // 최대 5개 동시 실행, 하나가 끝날 때마다 충원
  toArray,
);

함께 보기:

• p-limit/p-map에서 마이그레이션 — p-map과의 풀 의미론 비교(측정 포함)
• LLM 토큰 스트림 처리 — AsyncIterable 소스의 스트리밍 소비와 제한 팬아웃