비동기 작업의 동시성을 제한하는 간단한 패턴

대량 이미지 업로드나 알림 발송을 직렬로 처리하면 너무 느리고, 전부 동시에 처리하면 외부 시스템이 버티지 못한다. 필요한 것은 병렬 처리 여부를 고르는 것이 아니라 허용할 동시 작업 수를 정하는 것이다.

목차

• #작은 동시성 제한기 만들기
• #제한값을 정하는 기준
• #재시도와 함께 사용할 때
• #Promise를 미리 만들면 이미 늦다
• #실행 중인 개수를 그려 보면
• #실패 정책을 함수 옵션으로 드러내기
• #제한값은 어떻게 정할까
• #결론
• #관련 노트

작은 동시성 제한기 만들기

async function mapLimit<T, R>(
  items: T[],
  limit: number,
  worker: (item: T) => Promise<R>,
): Promise<R[]> {
  const results: R[] = new Array(items.length);
  let cursor = 0;

  async function run() {
    while (cursor < items.length) {
      const index = cursor++;
      results[index] = await worker(items[index]);
    }
  }

  await Promise.all(
    Array.from({ length: Math.min(limit, items.length) }, run),
  );
  return results;
}

worker를 limit개만 만들고 각 worker가 다음 항목을 가져간다. 결과 배열의 인덱스를 보존하므로 입력 순서와 결과 순서도 일치한다.

제한값을 정하는 기준

DB 작업이면 연결 풀보다 작게 시작한다. 외부 API라면 초당 요청 수와 응답 시간을 함께 본다. 메모리를 많이 쓰는 이미지 작업은 한 작업의 최대 메모리까지 측정한다.

재시도와 함께 사용할 때

실패한 작업이 제한기 밖에서 새 Promise를 만들면 순간 동시성이 다시 커질 수 있다. 재시도와 backoff를 worker 함수 내부에 넣어 한 슬롯이 작업의 전체 생명주기를 책임지게 한다.

Promise를 미리 만들면 이미 늦다

동시성 제한기를 사용할 때 흔히 하는 실수는 Promise 배열을 먼저 만든 뒤 제한기에 전달하는 것이다.

// 요청은 map을 실행하는 순간 모두 시작된다.
const requests = urls.map((url) => fetch(url));
const results = await runWithLimit(requests, 5);

제한기가 Promise를 전달받았을 때는 네트워크 요청이 이미 시작되었다. 제한하려면 아직 실행하지 않은 함수나 원본 입력을 전달해야 한다.

const tasks = urls.map((url) => () => fetch(url));
const results = await runTasksWithLimit(tasks, 5);

async function runTasksWithLimit(tasks, limit) {
  const results = new Array(tasks.length);
  let nextIndex = 0;

  async function worker() {
    while (true) {
      const currentIndex = nextIndex++;
      if (currentIndex >= tasks.length) return;
      results[currentIndex] = await tasks[currentIndex]();
    }
  }

  const workers = Array.from(
    { length: Math.min(limit, tasks.length) },
    () => worker(),
  );

  await Promise.all(workers);
  return results;
}

이 구현은 개념을 설명하기 위한 예시다. 작업 하나가 실패하면 worker의 Promise가 reject되고 Promise.all도 실패한다. 실제 요구에 따라 실패를 수집할지, 전체를 취소할지, 재시도할지를 추가해야 한다.

실행 중인 개수를 그려 보면

flowchart LR
    Q[대기 작업 20개] --> W1[worker 1]
    Q --> W2[worker 2]
    Q --> W3[worker 3]
    W1 --> R[결과 저장]
    W2 --> R
    W3 --> R

worker가 셋이라면 어느 순간에도 최대 세 작업만 실행된다. 한 worker가 작업을 끝내면 대기열의 다음 항목을 가져간다. 작업 시간이 서로 달라도 빈 worker가 다음 일을 가져가기 때문에 고정 청크 세 개를 순서대로 처리하는 것보다 자원 활용이 낫다.

실패 정책을 함수 옵션으로 드러내기

부분 실패를 허용하는 작업이라면 결과를 성공과 실패로 나눠 반환할 수 있다.

async function settleTask(task) {
  try {
    return { status: "fulfilled", value: await task() };
  } catch (reason) {
    return { status: "rejected", reason };
  }
}

const results = await runTasksWithLimit(
  tasks.map((task) => () => settleTask(task)),
  5,
);

반대로 첫 실패에서 중단해야 한다면 대기 중인 작업을 더 시작하지 않고, 이미 실행 중인 작업에는 공유 AbortSignal을 전달하는 구조가 필요하다.

제한값은 어떻게 정할까

처음부터 최댓값을 찾기보다 2, 4, 8처럼 늘려 보면서 처리량이 더 이상 증가하지 않거나 오류율이 올라가는 지점을 찾는다.

limit 2  -> 18 jobs/s, error 0.1%, p95 210ms
limit 4  -> 31 jobs/s, error 0.2%, p95 260ms
limit 8  -> 34 jobs/s, error 2.8%, p95 920ms

이 예시라면 8은 처리량 이득보다 지연과 오류 증가가 크다. 4를 기본값으로 두고 실제 트래픽에서 다시 관측하는 판단이 자연스럽다.

결론

좋은 동시성 값은 가장 큰 숫자가 아니라 하류 시스템을 안정적으로 유지하면서 목표 처리량을 내는 숫자다. 설정값으로 노출하고 운영 지표로 조정한다.

관련 노트

• 45장 프로미스
• Worker threads