Waylog Blog
← 목록으로 돌아가기

JavaScript AsyncIterator와 Web Streams API 실전: 백프레셔·취소·메모리 누수 없이 대용량 스트리밍 처리하기

JavaScript

ReadableStream과 AsyncIterator가 백프레셔 신호를 주고받는 다이어그램

콜백 지옥을 넘어선 자리에 또 다른 지옥이 있었다

우리 팀은 2026년 초, 실시간 AI 추론 결과를 클라이언트에 스트리밍하는 기능을 구축하면서 예상치 못한 장벽을 만났습니다. LLM API가 SSE(Server-Sent Events) 스트림으로 응답을 내려보내는데, 프론트엔드에서는 이를 onmessage 콜백으로 받아 상태 업데이트하고, 동시에 Node.js 중간 서버에서는 data 이벤트 핸들러를 중첩해 가공했습니다. 코드는 빠르게 스파게티가 됐고, 한 달이 지났을 때 RSS(Resident Set Size) 메모리가 배포 직후 280MB에서 4시간 만에 820MB까지 치솟는 현상이 재현됐습니다. Datadog 메모리 그래프는 완만한 계단식 상승을 보였고, 재시작 후 즉각 정상화되는 전형적인 참조 누수 패턴이었습니다.

원인은 단순했습니다. 요청이 취소됐는데도 이벤트 리스너가 정리되지 않았고, 각 리스너가 응답 본문 버퍼의 일부를 클로저로 붙들고 있었습니다. AsyncIterator와 Web Streams API로 전체 파이프라인을 재작성한 결과, 메모리 그래프는 수평선으로 안정됐고 코드 줄 수도 40% 줄었습니다. 이 글은 그 과정에서 익힌 설계 원칙과 실전 패턴을 공유합니다.

1. 왜 AsyncIterator가 콜백·Promise 체인보다 자연스러운가

콜백 기반 스트림 처리의 근본적인 문제는 제어 흐름의 역전(inversion of control) 입니다. 데이터가 언제 도착할지를 소비자가 아닌 생산자가 결정합니다. ondata 콜백이 호출될 때 소비자 코드가 아직 이전 데이터 처리를 마치지 못했다면, 버퍼에 쌓이거나 데이터가 유실됩니다. Promise 체인은 비동기 순서를 읽기 쉽게 만들었지만, 무한히 도착하는 데이터를 표현하기에는 설계 자체가 "단 한 번의 결과"에 맞춰져 있습니다.

AsyncIterator는 이 문제를 다르게 접근합니다. next() 호출 권한이 소비자에게 있습니다. 소비자가 준비됐을 때만 다음 값을 요청합니다. 이것이 백프레셔(backpressure) 의 본질입니다. 생산자가 아무리 빠르게 데이터를 만들어도, 소비자의 next() 호출이 없으면 기다려야 합니다.

MDN AsyncIterator 문서에 따르면, AsyncIterator 프로토콜은 { value, done } 형태의 Promise를 반환하는 next() 메서드를 가진 객체입니다. for await...of 루프는 이 프로토콜 위에 세워진 문법 설탕입니다.

콜백 대비 AsyncIterator의 장점은 명확합니다.

구분콜백/EventEmitterAsyncIterator
제어권생산자(push 방식)소비자(pull 방식)
백프레셔수동으로 pause()/resume() 구현next() 호출 속도가 자동 신호
취소리스너 수동 해제, 놓치기 쉬움return() 메서드 계약으로 정리
오류 전파error 이벤트 누락 시 uncaughtreject Promise로 자연스럽게 전파
합성어렵다for await...of 안에서 자연스럽게

단순함을 넘어, AsyncIterator는 스트리밍 데이터를 동기 배열처럼 다루는 인지적 단순함을 제공합니다. for 루프가 배열을 순회하듯, for await...of 가 비동기 스트림을 순회합니다. 에러는 try-catch로 잡고, 조기 종료는 break로 표현합니다. 이것이 코드베이스 가독성을 드라마틱하게 개선하는 이유입니다.

2. Symbol.asyncIterator 직접 구현하기 — 종료·예외 시 cleanup 계약

AsyncIterator 프로토콜의 핵심은 세 가지 메서드입니다. next()는 다음 값을 가져오고, return()은 소비자가 루프를 일찍 종료할 때 호출되며, throw()는 소비자가 에러를 이터레이터 안으로 주입할 때 사용합니다. return()과 throw()를 구현하지 않으면 파일 핸들, 네트워크 소켓, 이벤트 리스너가 청소되지 않습니다.

아래는 WebSocket 메시지를 AsyncIterator로 노출하는 실전 구현입니다. return() 안에서 소켓을 명시적으로 닫고 큐를 비워 누수를 방지합니다.

// websocket-async-iterator.ts
// WebSocket 메시지 스트림을 AsyncIterableIterator로 노출
// return()이 구현되어 있어 break/return 시 소켓이 즉시 정리됩니다

interface WsIteratorState<T> {
  queue: Array<(result: IteratorResult<T>) => void>;
  buffer: T[];
  done: boolean;
  error: unknown;
}

function createWebSocketAsyncIterator(
  url: string
): AsyncIterableIterator<MessageEvent<string>> {
  const state: WsIteratorState<MessageEvent<string>> = {
    queue: [],
    buffer: [],
    done: false,
    error: null,
  };

  const ws = new WebSocket(url);

  ws.addEventListener('message', (event: MessageEvent<string>) => {
    if (state.queue.length > 0) {
      // 소비자가 이미 next()를 호출해서 대기 중 — 즉시 resolve
      state.queue.shift()!({ value: event, done: false });
    } else {
      // 소비자가 아직 next()를 부르지 않음 — 버퍼에 적재
      state.buffer.push(event);
    }
  });

  ws.addEventListener('close', () => {
    state.done = true;
    for (const resolve of state.queue) {
      resolve({ value: undefined as any, done: true });
    }
    state.queue.length = 0;
  });

  ws.addEventListener('error', (e) => {
    state.error = e;
    state.done = true;
    for (const resolve of state.queue) {
      resolve({ value: undefined as any, done: true });
    }
    state.queue.length = 0;
  });

  const iterator: AsyncIterableIterator<MessageEvent<string>> = {
    [Symbol.asyncIterator]() {
      return this;
    },

    next(): Promise<IteratorResult<MessageEvent<string>>> {
      if (state.buffer.length > 0) {
        return Promise.resolve({ value: state.buffer.shift()!, done: false });
      }
      if (state.done) {
        if (state.error) return Promise.reject(state.error);
        return Promise.resolve({ value: undefined as any, done: true });
      }
      // 버퍼가 비고 아직 완료되지 않음 — 큐에 등록하고 대기
      return new Promise((resolve) => {
        state.queue.push(resolve);
      });
    },

    // cleanup 계약: for await...of 가 break/return으로 일찍 종료될 때 런타임이 자동 호출
    return(): Promise<IteratorResult<MessageEvent<string>>> {
      state.done = true;
      state.buffer.length = 0;
      if (
        ws.readyState === WebSocket.OPEN ||
        ws.readyState === WebSocket.CONNECTING
      ) {
        ws.close(1000, 'Iterator returned early');
      }
      return Promise.resolve({ value: undefined as any, done: true });
    },
  };

  return iterator;
}

// 사용 예시: 가격이 10만 원을 넘으면 자동 구독 해제
async function watchQuotes(symbol: string) {
  const iter = createWebSocketAsyncIterator(
    `wss://stream.example.com/quotes/${symbol}`
  );

  for await (const msg of iter) {
    const quote = JSON.parse(msg.data) as { symbol: string; price: number };
    console.log(`${quote.symbol}: ${quote.price.toLocaleString()}원`);

    // break 시 return()이 자동 호출 → WebSocket 즉시 정리
    if (quote.price > 100_000) break;
  }
  // 이 시점에서 WebSocket은 이미 닫혀 있습니다
}

이 구현에서 가장 중요한 부분은 return() 메서드입니다. for await...of 루프가 break, return, 예외 전파로 중단되면 JavaScript 런타임이 이터레이터의 return() 메서드를 자동으로 호출합니다. 이것이 AsyncIterator의 cleanup 계약이며, 콜백 기반에서는 개발자가 모두 직접 챙겨야 했던 부분입니다.

3. for await...of의 직렬성과 동시성 트레이드오프

for await...of는 직렬(serial) 로 동작합니다. next()를 호출하고, 반환된 Promise가 resolve될 때까지 다음 next() 호출을 보류합니다. 이 특성은 백프레셔 구현에 이상적이지만, 동시에 처리할 수 있는 여러 비동기 작업을 순차적으로 처리하게 만드는 성능 병목이 될 수 있습니다.

10개의 URL에서 데이터를 fetch해야 할 때, for await...of로 순서대로 처리하면 총 시간은 개별 fetch 시간의 합이 됩니다. 이때 Promise.all을 사용하면 동시성을 회복할 수 있습니다.

// 직렬 처리 (느림): 각 fetch가 완료된 후 다음 fetch 시작
async function fetchSequential(urls: string[]): Promise<string[]> {
  const results: string[] = [];
  for (const url of urls) {
    const res = await fetch(url);
    results.push(await res.text());
  }
  return results;
}

// 병렬 처리 (빠름): 모든 fetch를 동시에 시작, 결과 순서 보장
async function fetchParallel(urls: string[]): Promise<string[]> {
  return Promise.all(urls.map(async (url) => {
    const res = await fetch(url);
    return res.text();
  }));
}

// 슬라이딩 윈도우 패턴 (균형): 동시 실행 수를 N개로 제한
async function fetchWithConcurrencyLimit(
  urls: string[],
  maxConcurrent: number
): Promise<string[]> {
  const results: string[] = new Array(urls.length);
  const executing = new Set<Promise<void>>();

  for (let i = 0; i < urls.length; i++) {
    const idx = i;
    const promise: Promise<void> = fetch(urls[idx])
      .then((res) => res.text())
      .then((text) => {
        results[idx] = text;
        executing.delete(promise);
      });

    executing.add(promise);

    if (executing.size >= maxConcurrent) {
      // 가장 빨리 완료되는 것을 기다린 후 다음 요청 시작
      await Promise.race(executing);
    }
  }

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

Promise.all 활용 시 주의할 점이 있습니다. 하나의 Promise가 reject되면 다른 Promise들이 취소 없이 계속 실행됩니다. 대용량 스트림 처리에서는 Promise.allSettled나 명시적 AbortController를 통한 취소 신호가 필요합니다. Node.js Worker Threads와 CPU 집약적 작업에서도 동시성 제어와 취소 패턴을 상세히 다루고 있습니다.

4. Web Streams API 기초: Readable / Writable / Transform

WHATWG Streams 명세는 세 종류의 스트림을 정의합니다. ReadableStream은 데이터를 생산하고, WritableStream은 데이터를 소비하며, TransformStream은 생산과 소비를 동시에 수행하는 변환기입니다.

MDN Streams API 문서에 따르면, Web Streams는 2016년부터 브라우저에 등장했으며 2026년 현재 Chrome, Firefox, Safari, Edge 전 최신 버전에서 완전히 지원됩니다. Node.js에서는 v18부터 글로벌 객체로 사용할 수 있습니다.

ReadableStream의 생성자는 underlyingSource 객체를 받습니다. 이 객체의 pull 메서드가 백프레셔의 핵심입니다. pull은 소비자가 더 많은 데이터를 원할 때, 정확히는 내부 큐의 desiredSize가 0 이상일 때만 호출됩니다. 생산자가 데이터를 밀어넣는 push 방식이 아니라, 소비자가 당기는 pull 방식입니다.

WritableStream의 underlyingSink는 write, close, abort 핸들러를 받습니다. write가 Promise를 반환하면, 해당 Promise가 settle될 때까지 다음 write 호출이 보류됩니다. 이것이 WritableStream 쪽의 백프레셔 구현입니다.

TransformStream은 writable과 readable 두 개의 스트림을 연결합니다. transform 핸들러에서 controller.enqueue()로 변환된 값을 readable 쪽으로 내보내고, controller.error()로 에러를 전파합니다.

5. 백프레셔는 어떻게 흐르는가 — desiredSize / pull / queuingStrategy

백프레셔는 WHATWG Streams 명세에서 가장 정교하게 설계된 부분입니다. ReadableStream 내부에는 내부 큐(internal queue) 가 있고, 이 큐가 얼마나 차 있는지를 desiredSize로 나타냅니다.

  • desiredSize > 0: 큐에 여유가 있음. 생산자는 enqueue해도 됩니다.
  • desiredSize === 0: 큐가 가득 참. pull이 호출되지 않습니다.
  • desiredSize < 0: 큐가 초과됨. 생산자는 즉시 멈춰야 합니다.

desiredSize가 0 이하일 때 pull은 호출되지 않습니다. 생산자는 자동으로 데이터 생산을 멈춥니다. 이것이 자동 백프레셔이고, 수동 pause()/resume() 없이 작동합니다.

queuingStrategy는 큐 용량을 어떻게 계산할지를 정의합니다.

전략highWaterMark 단위적합한 상황
CountQueuingStrategy청크 개수청크 크기가 균일한 객체 스트림
ByteLengthQueuingStrategy바이트 수HTTP 응답 등 가변 크기 청크
커스텀 strategy자유 정의우선순위 큐, 가중치 기반 큐 등

highWaterMark는 큐가 "가득 찼다"고 판단하는 임계값입니다. ByteLengthQueuingStrategy({ highWaterMark: 64 * 1024 })는 64KB가 쌓이면 백프레셔를 발동시킵니다. 너무 낮으면 불필요한 pull 억제로 처리량이 떨어지고, 너무 높으면 메모리를 낭비합니다. 대부분의 HTTP 응답 처리 케이스에서는 16KB~64KB 범위가 적절합니다.

TransformStream의 백프레셔는 writable 쪽의 desiredSize가 readable 쪽의 pull 호출 여부에 영향을 줍니다. 다운스트림 소비자가 느리면 업스트림 생산자도 자동으로 느려집니다. 파이프라인 전체가 가장 느린 단계의 속도에 맞춰 흐릅니다.

6. pipeTo·pipeThrough·tee로 합성 가능한 파이프라인 만들기

Web Streams의 합성 API는 세 가지입니다.

readable.pipeTo(writable, options): ReadableStream의 모든 데이터를 WritableStream으로 이송하고, 완료 또는 에러 시 양쪽 스트림을 정리합니다. options.signal로 AbortSignal을 연결하면 취소 가능합니다.

readable.pipeThrough(transform, options): ReadableStream을 TransformStream의 writable 쪽에 연결하고, transform의 readable 쪽을 반환합니다. 메서드 체이닝으로 여러 변환을 순서대로 연결할 수 있습니다.

readable.tee(): 단일 ReadableStream을 두 개의 독립적인 ReadableStream으로 복제합니다.

tee()의 함정은 느린 소비자가 있으면 메모리가 무제한으로 증가할 수 있다는 점입니다. 예를 들어 로깅용 스트림과 처리용 스트림으로 tee()를 나눴을 때, 처리용 스트림이 느리면 로깅이 이미 읽은 데이터도 처리용 소비자를 위해 버퍼에 남아야 합니다. Edge Runtime 비교 — Cloudflare Workers vs Vercel Edge에서 다룬 것처럼, Edge 환경에서는 tee() 사용 시 메모리 제한(Cloudflare Workers 128MB)에 특히 주의해야 합니다.

tee() 대신 TransformStream 안에서 명시적으로 분기하거나, 느린 소비자에게는 샘플링된 데이터만 전달하는 전략을 권장합니다.

7. AbortController로 안전하게 취소·정리하기

fetch와 Web Streams API는 모두 AbortSignal을 직접 지원합니다. pipeTo와 pipeThrough는 두 번째 인자로 { signal } 옵션을 받습니다. signal이 중단되면 파이프라인 전체가 즉시 정리됩니다.

취소 패턴에서 흔히 발생하는 실수는 AbortController를 너무 많이 생성하거나 생명주기가 다른 신호들을 하나의 컨트롤러로 합치는 것입니다. 요청 단위와 파이프라인 단위의 취소를 분리해야 하는 경우, AbortSignal.any([signal1, signal2])로 여러 신호를 합성합니다. 2026년 현재 Chrome 116+, Firefox 115+, Safari 17.4+에서 지원됩니다.

타임아웃은 AbortSignal.timeout(milliseconds) 정적 메서드로 간결하게 처리합니다. 예를 들어 사용자 취소와 자동 타임아웃을 합쳐 React 컴포넌트 언마운트 시 자동 정리되는 패턴을 만들 수 있습니다. useEffect의 cleanup 함수에서 controller.abort()만 호출하면 모든 fetch와 스트림이 즉시 정리됩니다.

잠금된 리더(locked reader) 문제도 주의해야 합니다. ReadableStream은 한 번에 하나의 리더만 가질 수 있고, pipeTo나 pipeThrough가 스트림을 잠그면 다른 리더를 연결할 수 없습니다. stream.locked 프로퍼티로 잠금 여부를 확인할 수 있습니다. 이미 잠긴 스트림에 getReader()를 호출하면 TypeError: ReadableStream is locked 예외가 발생합니다.

8. Node.js Streams ↔ Web Streams 상호 운용

Node.js 공식 Streams 문서에 따르면, Node.js v16.7.0부터 stream.Readable.toWeb(), stream.Writable.toWeb(), stream.Readable.fromWeb() API가 추가됐습니다. 2026년 현재 Node.js 20 이상에서는 이 API들이 stable 상태입니다.

이 상호 운용 레이어는 Node.js의 기존 스트림 생태계(예: fs.createReadStream, zlib.createGzip, csv-parse)와 브라우저 호환 Web Streams API를 연결하는 다리 역할을 합니다.

변환 방향API
Node Readable → Web ReadableStreamReadable.toWeb(nodeReadable)
Web ReadableStream → Node ReadableReadable.fromWeb(webReadableStream)
Node Writable → Web WritableStreamWritable.toWeb(nodeWritable)
Node Duplex → Web TransformStreamDuplex.toWeb(nodeDuplex)

주의사항: Readable.fromWeb()으로 변환된 Node.js 스트림은 objectMode가 아닌 경우 청크를 Buffer로 받습니다. 파이프라인 중간에서 인코딩 변환이 필요하면 TextDecoder/TextEncoder 기반 TransformStream을 삽입해야 합니다. 또한 Node.js Readable의 pause()/resume() 기반 백프레셔와 Web Streams의 desiredSize 기반 백프레셔는 서로 다른 메커니즘이므로, 변환 레이어를 거치면 두 시스템의 백프레셔 신호가 100% 일치하지 않을 수 있습니다. 대용량 처리에서는 변환 레이어 양쪽의 메모리 사용량을 별도로 모니터링하는 것을 권장합니다.

9. 실전: Fetch SSE를 AsyncIterator로 노출하는 라이브러리 패턴

LLM API(OpenAI, Anthropic 등)의 스트리밍 응답은 text/event-stream 형식인 SSE(Server-Sent Events)입니다. 브라우저의 EventSource는 GET만 지원하고 커스텀 헤더를 붙일 수 없어 LLM API 인증에 사용하기 어렵습니다. 때문에 fetch로 직접 응답 본문을 스트리밍하고, AsyncIterator로 노출하는 패턴이 실무에서 표준처럼 자리잡았습니다.

우리 팀이 내부에서 사용하는 SSE 클라이언트 유틸리티 패턴입니다. ReadableStream을 Web Streams TransformStream으로 파싱하고, 최종 소비자는 단순한 for await...of 루프로 토큰을 받아 화면에 표시합니다.

// fetch-sse-iterator.ts
// Server-Sent Events 응답을 AsyncGenerator<SseEvent>로 노출

export interface SseEvent {
  id?: string;
  event: string;
  data: string;
}

/** SSE 라인들을 완성된 SseEvent 객체로 조립하는 TransformStream */
function createSseParser(): TransformStream<string, SseEvent> {
  let id: string | undefined;
  let event = 'message';
  let dataLines: string[] = [];

  return new TransformStream<string, SseEvent>({
    transform(line, controller) {
      // 빈 줄 = 이벤트 경계
      if (line === '') {
        if (dataLines.length > 0) {
          controller.enqueue({
            id,
            event,
            data: dataLines.join('\n'),
          });
        }
        event = 'message';
        dataLines = [];
        return;
      }

      if (line.startsWith('data: ')) {
        dataLines.push(line.slice(6));
      } else if (line.startsWith('id: ')) {
        id = line.slice(4);
      } else if (line.startsWith('event: ')) {
        event = line.slice(7);
      }
    },
  });
}

/** Uint8Array 청크를 줄 단위 string으로 분해하는 TransformStream */
function createLineDecoder(): TransformStream<Uint8Array, string> {
  const decoder = new TextDecoder('utf-8');
  let buffer = '';

  return new TransformStream<Uint8Array, string>({
    transform(chunk, controller) {
      buffer += decoder.decode(chunk, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop() ?? '';
      for (const line of lines) {
        controller.enqueue(line.replace(/\r$/, ''));
      }
    },
    flush(controller) {
      if (buffer) controller.enqueue(buffer.replace(/\r$/, ''));
    },
  });
}

/**
 * Fetch SSE 스트림을 AsyncGenerator로 소비하는 공개 API
 */
export async function* fetchSseStream(
  url: string,
  init: RequestInit = {}
): AsyncGenerator<SseEvent> {
  const response = await fetch(url, {
    ...init,
    headers: {
      Accept: 'text/event-stream',
      'Cache-Control': 'no-cache',
      ...init.headers,
    },
  });

  if (!response.ok) {
    const body = await response.text().catch(() => '');
    throw new Error(`SSE fetch 실패 (${response.status}): ${body.slice(0, 200)}`);
  }

  if (!response.body) {
    throw new Error('응답 본문이 없습니다 (response.body is null)');
  }

  const sseStream = response.body
    .pipeThrough(createLineDecoder())
    .pipeThrough(createSseParser());

  const reader = sseStream.getReader();
  try {
    while (true) {
      const { done, value } = await reader.read();
      if (done) break;
      if (value.data === '[DONE]') return;
      yield value;
    }
  } finally {
    reader.releaseLock();
  }
}

이 패턴은 JavaScript Signals와 세밀한 반응성과 결합할 때 특히 강력합니다. LLM 토큰이 도착할 때마다 Signal 값을 갱신하면, 구독하는 컴포넌트만 정확히 리렌더링됩니다. SSE 스트림을 Signal로 브리지하는 useStreamingSignal() 훅 패턴은 실시간 AI 인터페이스에서 매우 효과적입니다.

10. 결론: 누수·잠금·교착 디버깅 체크리스트

프로덕션에서 스트림 기반 코드를 운영하면서 우리 팀이 반복적으로 마주친 문제들을 정리합니다. 이 체크리스트는 코드 리뷰와 인시던트 대응 모두에서 사용합니다.

메모리 누수 징후: RSS 메모리가 배포 직후부터 완만하게 계단식으로 상승하다가 재시작 후 즉각 정상화되면 스트림 또는 이터레이터의 cleanup 누락을 의심합니다. Chrome DevTools Memory 탭의 "Detached" 노드, Node.js의 --expose-gc + global.gc() + process.memoryUsage() 추적으로 누수 출처를 확인합니다.

잠금 오류 TypeError: ReadableStream is locked: 하나의 ReadableStream에 두 번 이상 getReader(), pipeTo(), pipeThrough()를 호출할 때 발생합니다. stream.locked 프로퍼티를 먼저 확인하고, 여러 소비자가 필요하다면 tee()로 분기합니다.

교착(deadlock) 패턴: TransformStream의 transform 핸들러 안에서 자기 자신의 readable 쪽을 동기적으로 읽으려 하거나, WritableStream의 write 핸들러가 무한 대기하는 Promise를 반환하면 파이프라인 전체가 멈춥니다.

실무 체크리스트:

  1. AsyncIterator를 직접 구현할 때 return() 메서드를 반드시 작성한다. for await...of가 break나 예외로 종료될 때 런타임이 이를 자동 호출하므로, 소켓·핸들·리스너 정리 코드를 이 안에 배치한다.

  2. ReadableStream은 단 하나의 소비자만 가질 수 있음을 기억한다. 여러 소비자가 필요하면 tee()를 사용하되, 소비 속도 차이가 크면 느린 쪽 버퍼를 별도로 모니터링한다.

  3. AbortController를 요청 생명주기와 일대일로 연결한다. 컴포넌트 언마운트, 탭 닫힘, 자동 타임아웃 등 여러 취소 트리거를 AbortSignal.any()로 합성하고, 각 파이프라인 단계에 동일한 signal을 전달한다.

  4. Node.js와 Web Streams를 혼용할 때 Readable.toWeb() / Readable.fromWeb() 변환 레이어를 명시적으로 삽입한다. 두 API의 백프레셔 메커니즘이 다르므로, 변환 경계에서 메모리 사용량을 측정한다.

  5. highWaterMark 값을 시나리오별로 튜닝한다. 기본값(1 청크)은 처리량이 낮고, 너무 크면 메모리를 낭비한다. 실제 청크 크기와 처리 지연을 기준으로 16KB~64KB 범위에서 시작하고, 벤치마크 후 조정한다.

참고 자료