junyeokk
Blog
Architecture·2025. 11. 21

Producer-Consumer 패턴

서버가 사용자 요청을 받을 때마다 이메일을 보내거나, 이미지를 리사이징하거나, 외부 API를 호출해야 하는 상황을 생각해 보자. 이런 작업은 수백 밀리초에서 수 초까지 걸릴 수 있는데, 사용자는 그 결과를 즉시 확인할 필요가 없다. 그런데도 요청 핸들러 안에서 이 작업을 동기적으로 처리하면 응답 시간이 느려지고, 작업이 실패했을 때 전체 요청이 실패한다.

Producer-Consumer 패턴은 이 문제를 작업을 "요청하는 쪽"과 "처리하는 쪽"을 분리함으로써 해결한다. Producer는 작업을 큐에 넣기만 하고 즉시 반환한다. Consumer는 큐에서 작업을 꺼내서 독립적으로 처리한다. 이 두 쪽은 서로를 직접 알지 못하고, 오직 큐를 통해서만 소통한다.

동기 처리 vs Producer-Consumer

동기 처리

Client → Server: POST /signup Server: 1. DB에 유저 저장 (50ms) 2. 환영 이메일 발송 (800ms) ← 여기서 오래 걸림 3. 슬랙 알림 전송 (200ms) ← 실패하면? Server → Client: 201 Created (총 1050ms)

이메일 서버가 응답을 안 하면? 슬랙이 일시적으로 다운되면? 사용자의 회원가입 자체가 실패하거나, 타임아웃이 발생한다. 핵심 비즈니스 로직(유저 생성)이 부가 작업(알림 발송)에 의존하게 되는 것이다.

Producer-Consumer 분리

Client → Server: POST /signup Server: 1. DB에 유저 저장 (50ms) 2. 큐에 "이메일 발송" 작업 추가 (5ms) 3. 큐에 "슬랙 알림" 작업 추가 (5ms) Server → Client: 201 Created (총 60ms) --- 별도 프로세스 --- Email Worker: 큐에서 작업 꺼내서 이메일 발송 Slack Worker: 큐에서 작업 꺼내서 슬랙 알림 전송

응답 시간이 1050ms에서 60ms로 줄었다. 이메일 서버가 다운되어도 회원가입은 성공하고, 이메일 발송은 나중에 재시도하면 된다.

핵심 구성 요소

Producer (생산자)

작업을 생성해서 큐에 넣는 역할이다. 작업의 처리 결과에 관심이 없고, 큐에 성공적으로 넣었는지만 확인한다.

typescript
Client → Server: POST /signup
Server: 1. DB에 유저 저장 (50ms)
        2. 환영 이메일 발송 (800ms)  ← 여기서 오래 걸림
        3. 슬랙 알림 전송 (200ms)    ← 실패하면?
Server → Client: 201 Created (총 1050ms)

persistent: true를 설정하면 메시지가 디스크에 저장되어 브로커가 재시작되어도 메시지가 유실되지 않는다. 프로덕션 환경에서는 거의 항상 활성화해야 한다.

Queue (큐)

Producer와 Consumer 사이의 버퍼다. 메시지 브로커(RabbitMQ, Redis, Kafka 등)가 이 역할을 수행한다. 큐가 제공하는 핵심 보장은 다음과 같다.

  • 순서 보장: 먼저 들어온 메시지가 먼저 처리된다 (FIFO)
  • 영속성: 브로커가 재시작되어도 메시지가 유실되지 않는다 (설정에 따라)
  • 부하 분산: 여러 Consumer가 있으면 메시지가 분산된다

Consumer (소비자)

큐에서 메시지를 꺼내서 실제 작업을 처리한다. Producer와 독립적으로 실행되는 별도의 프로세스다.

typescript
Client → Server: POST /signup
Server: 1. DB에 유저 저장 (50ms)
        2. 큐에 "이메일 발송" 작업 추가 (5ms)
        3. 큐에 "슬랙 알림" 작업 추가 (5ms)
Server → Client: 201 Created (총 60ms)

--- 별도 프로세스 ---
Email Worker: 큐에서 작업 꺼내서 이메일 발송
Slack Worker: 큐에서 작업 꺼내서 슬랙 알림 전송

여기서 핵심은 ack/nack 메커니즘이다. Consumer가 메시지를 받았다고 해서 큐에서 바로 삭제되는 게 아니다. ack()를 호출해야 비로소 큐에서 제거된다. 처리 중 에러가 발생하면 nack()로 메시지를 다시 큐에 넣어서 재시도할 수 있다.

ack/nack 전략

메시지 확인(acknowledgment)은 Producer-Consumer 패턴에서 가장 중요한 부분이다. 잘못 설정하면 메시지가 유실되거나 무한 재시도에 빠질 수 있다.

Auto Ack (자동 확인)

typescript
class EmailProducer {
  private channel: Channel;

  constructor(channel: Channel) {
    this.channel = channel;
  }

  async sendWelcomeEmail(userId: string, email: string): Promise<void> {
    const message = {
      type: 'welcome',
      userId,
      email,
      createdAt: new Date().toISOString(),
    };

    this.channel.publish(
      'notification-exchange',
      'email.welcome',
      Buffer.from(JSON.stringify(message)),
      { persistent: true }
    );
  }
}

메시지를 받는 즉시 큐에서 제거된다. 처리 중 에러가 발생하면 메시지가 사라진다. 로그 수집처럼 일부 유실이 허용되는 경우에만 사용한다.

Manual Ack (수동 확인)

typescript
class EmailConsumer {
  private channel: Channel;
  private mailer: MailerService;

  async start(): Promise<void> {
    await this.channel.consume('email-queue', async (msg) => {
      if (!msg) return;

      try {
        const data = JSON.parse(msg.content.toString());
        await this.mailer.send(data.email, data.type);
        this.channel.ack(msg);  // 처리 완료 확인
      } catch (error) {
        this.channel.nack(msg, false, true);  // 재시도를 위해 큐에 다시 넣기
      }
    });
  }
}

대부분의 프로덕션 환경에서는 Manual Ack를 사용한다. nack()의 세 번째 인자 requeue를 true로 설정하면 메시지가 큐의 맨 앞에 다시 들어간다. false로 설정하면 Dead Letter Exchange로 보내거나 버린다.

주의할 점

nack(msg, false, true)로 무조건 재시도하면 처리할 수 없는 메시지(잘못된 JSON, 존재하지 않는 유저 등)가 무한 루프에 빠진다. 이런 메시지를 Poison Message라고 부른다. 이를 방지하려면 재시도 횟수를 제한하거나, Dead Letter Queue로 보내서 별도 처리해야 한다.

typescript
channel.consume('queue', callback, { noAck: true });

독립 프로세스로 분리하기

Producer-Consumer 패턴의 진짜 가치는 Consumer를 별도의 프로세스로 분리할 때 나타난다. 같은 프로세스 안에서 Producer와 Consumer를 함께 실행하면 결국 CPU와 메모리를 공유하게 되어 분리의 의미가 줄어든다.

왜 별도 프로세스인가?

[API 서버 프로세스] [이메일 워커 프로세스] ├─ HTTP 요청 처리 ├─ 큐에서 메시지 소비 ├─ 비즈니스 로직 ├─ SMTP로 이메일 발송 └─ 큐에 메시지 발행 └─ 성공/실패 처리

이렇게 분리하면:

  • 독립적 스케일링: 이메일 발송이 밀리면 워커 인스턴스만 늘리면 된다. API 서버를 함께 늘릴 필요가 없다.
  • 격리된 장애: 이메일 워커가 크래시되어도 API 서버는 정상 작동한다.
  • 독립적 배포: 이메일 템플릿이 바뀌어서 워커를 재배포해도 API 서버에 영향이 없다.
  • 리소스 최적화: CPU 집약적인 작업(이미지 리사이징)을 별도 프로세스로 분리하면 API 서버의 응답 시간에 영향을 주지 않는다.

워커 진입점

독립 프로세스는 자체 진입점(entrypoint)이 필요하다. 간단한 예시:

typescript
channel.consume('queue', async (msg) => {
  try {
    await processMessage(msg);
    channel.ack(msg);           // 성공: 큐에서 제거
  } catch (error) {
    channel.nack(msg, false, true);  // 실패: 큐에 다시 넣기
  }
}, { noAck: false });

prefetch(1)은 한 번에 하나의 메시지만 가져오도록 제한한다. Consumer가 현재 메시지를 ack하기 전까지 새로운 메시지를 받지 않는다. 여러 워커가 동시에 실행되면 라운드 로빈으로 메시지가 분배된다.

Prefetch와 동시성 제어

prefetch는 Consumer가 한 번에 처리할 수 있는 메시지의 최대 수를 제한한다. 이 값을 어떻게 설정하느냐에 따라 처리 속도와 안정성이 크게 달라진다.

typescript
channel.consume('queue', async (msg) => {
  const headers = msg.properties.headers || {};
  const retryCount = headers['x-retry-count'] || 0;

  try {
    await processMessage(msg);
    channel.ack(msg);
  } catch (error) {
    if (retryCount >= 3) {
      // 3번 실패하면 Dead Letter Queue로
      channel.nack(msg, false, false);
    } else {
      // 재시도 횟수를 증가시켜서 다시 발행
      channel.ack(msg);
      channel.publish('exchange', 'routing.key',
        msg.content,
        { headers: { 'x-retry-count': retryCount + 1 } }
      );
    }
  }
});
prefetch장점단점
1공정한 분배, 실패 시 재처리 최소네트워크 왕복으로 느림
5~20처리량과 안정성의 균형실패 시 여러 메시지 재처리
0 (무제한)최대 처리량메모리 폭발, 불공정 분배

일반적으로 이메일 발송 같은 I/O 바운드 작업은 5~10, 이미지 처리 같은 CPU 바운드 작업은 1~2로 설정한다.

복수 Consumer와 경쟁 소비

하나의 큐에 여러 Consumer를 연결하면 경쟁 소비(Competing Consumers) 패턴이 된다. 메시지가 여러 Consumer에 라운드 로빈으로 분배되어 병렬 처리가 가능하다.

Producer → [Queue] → Consumer 1 (msg1, msg4, msg7...) → Consumer 2 (msg2, msg5, msg8...) → Consumer 3 (msg3, msg6, msg9...)

이 패턴의 핵심 보장은 하나의 메시지는 정확히 하나의 Consumer만 처리한다는 것이다. 메시지가 중복 처리되지 않으므로 이메일이 두 번 발송되는 일은 없다 (단, Consumer가 처리 중 크래시되면 메시지가 다시 큐에 들어가서 다른 Consumer가 처리할 수 있다. 이 경우 멱등성이 중요하다).

수평 확장

트래픽이 증가하면 Consumer 인스턴스를 늘리기만 하면 된다.

bash
[API 서버 프로세스]        [이메일 워커 프로세스]
 ├─ HTTP 요청 처리          ├─ 큐에서 메시지 소비
 ├─ 비즈니스 로직           ├─ SMTP로 이메일 발송
 └─ 큐에 메시지 발행        └─ 성공/실패 처리

컨테이너 환경에서는 replica 수를 조절한다.

yaml
// worker/main.ts
async function bootstrap() {
  // 1. 설정 로드
  const config = loadConfig();

  // 2. 메시지 브로커 연결
  const connection = await connect(config.amqpUrl);
  const channel = await connection.createChannel();

  // 3. 큐 선언
  await channel.assertQueue('email-queue', { durable: true });

  // 4. prefetch 설정
  channel.prefetch(1);

  // 5. Consumer 시작
  const consumer = new EmailConsumer(channel, mailerService);
  await consumer.start();

  console.log('Email worker started');
}

bootstrap().catch(console.error);

큐에 메시지가 쌓이는 속도가 Consumer의 처리 속도보다 빠르면 Consumer를 늘리고, 큐가 거의 비어 있으면 줄인다. 이런 오토스케일링을 큐 깊이(queue depth) 기반으로 자동화할 수도 있다.

메시지 설계 원칙

큐에 넣는 메시지를 어떻게 설계하느냐도 중요하다.

충분한 정보를 담아라

typescript
// 한 번에 1개만 처리 (안전하지만 느림)
channel.prefetch(1);

// 한 번에 10개 처리 (빠르지만 실패 시 10개 재처리)
channel.prefetch(10);

// 제한 없음 (위험! 메모리 폭발 가능)
channel.prefetch(0);

Consumer가 메시지만 보고 작업을 완료할 수 있어야 한다. 다시 DB를 조회해야 한다면 Producer와 Consumer 사이에 불필요한 의존성이 생긴다.

멱등성을 보장해라

네트워크 문제로 같은 메시지가 두 번 처리될 수 있다. Consumer는 같은 메시지를 여러 번 처리해도 결과가 같도록 설계해야 한다.

typescript
Producer → [Queue] → Consumer 1 (msg1, msg4, msg7...)
                   → Consumer 2 (msg2, msg5, msg8...)
                   → Consumer 3 (msg3, msg6, msg9...)

버전 관리

메시지 형식이 바뀔 수 있으므로 버전 필드를 넣어두면 하위 호환이 쉽다.

typescript
# 이메일 워커 3개 실행
node dist/worker/main.js &
node dist/worker/main.js &
node dist/worker/main.js &

에러 처리와 Dead Letter Queue

실패한 메시지를 어떻게 처리할지는 시스템의 안정성을 결정한다.

일시적 에러 vs 영구적 에러

typescript
# docker-compose.yml
services:
  email-worker:
    build: .
    command: node dist/worker/main.js
    deploy:
      replicas: 3

일시적 에러(네트워크 문제, rate limit)는 재시도하면 해결될 가능성이 높다. 영구적 에러(잘못된 데이터, 존재하지 않는 사용자)는 몇 번을 재시도해도 실패한다. 이 둘을 구분하지 않으면 처리 불가능한 메시지가 큐를 점유해서 정상 메시지의 처리를 방해한다.

Dead Letter Queue (DLQ)

영구적으로 실패한 메시지는 Dead Letter Queue로 보낸다. DLQ는 "처리 불가능한 메시지의 무덤"이 아니라, 나중에 분석하고 수동으로 재처리할 수 있는 대기열이다.

typescript
// ❌ 나쁜 예: Consumer가 DB를 다시 조회해야 한다
{ type: 'welcome', userId: '123' }

// ✅ 좋은 예: Consumer가 독립적으로 처리할 수 있다
{
  type: 'welcome',
  userId: '123',
  email: 'user@example.com',
  nickname: 'John',
  locale: 'ko',
  createdAt: '2025-01-15T09:00:00Z'
}

DLQ에 쌓인 메시지를 모니터링하면 시스템의 문제를 빠르게 파악할 수 있다. DLQ에 메시지가 급격히 늘어나면 외부 서비스 장애나 코드 버그를 의심할 수 있다.

모니터링

Producer-Consumer 시스템에서 반드시 모니터링해야 하는 메트릭:

메트릭의미경고 조건
Queue Depth큐에 쌓인 메시지 수지속적으로 증가
Consumer Count활성 Consumer 수0이 되면 위험
Publish Rate초당 발행 메시지 수비정상적 급증
Consume Rate초당 소비 메시지 수Publish Rate보다 낮으면
DLQ DepthDead Letter Queue 깊이0보다 크면 확인 필요

Queue Depth가 계속 증가하면 Consumer가 부족하거나 처리 속도가 느리다는 뜻이다. Consumer Count가 0이 되면 메시지가 무한정 쌓이게 되므로 즉시 대응해야 한다.

다른 비동기 처리 방식과 비교

인메모리 큐 (Bull, BullMQ)

typescript
async processMessage(data: EmailMessage): Promise<void> {
  // 이미 발송된 이메일인지 확인
  const alreadySent = await this.checkIfSent(data.userId, data.type);
  if (alreadySent) {
    return;  // 중복이면 스킵
  }

  await this.mailer.send(data.email, data.type);
  await this.markAsSent(data.userId, data.type);
}

인메모리 큐는 설정이 간단하고 같은 언어 생태계 안에서 사용하기 편하다. 하지만 Redis가 다운되면 메시지가 유실될 수 있고, 다른 언어로 작성된 Consumer와 연동하기 어렵다.

이벤트 기반 (EventEmitter)

typescript
{
  version: 2,
  type: 'welcome',
  // ...
}

같은 프로세스 내에서 모듈 간 통신에 적합하다. 하지만 프로세스가 죽으면 이벤트가 사라지고, 재시도 메커니즘이 없고, 프로세스 간 통신이 불가능하다.

비교 정리

방식프로세스 분리영속성재시도스케일링복잡도
EventEmitter✗✗✗✗낮음
BullMQ (Redis)○△✓○중간
RabbitMQ✓✓✓✓높음
Kafka✓✓✓✓매우 높음

간단한 백그라운드 작업이면 BullMQ로 충분하다. 프로세스를 분리하고 언어에 구애받지 않는 안정적인 메시지 전달이 필요하면 RabbitMQ가 적합하다. 대용량 이벤트 스트리밍이 필요하면 Kafka를 고려한다.

정리

  • Producer는 큐에 넣고 즉시 반환하고, Consumer는 별도 프로세스에서 독립적으로 처리하며, ack/nack 수동 확인이 메시지 유실을 방지한다
  • 일시적 에러는 재시도하고 영구적 에러는 Dead Letter Queue로 보내되, 재시도 횟수를 제한해서 Poison Message 무한 루프를 막는다
  • 메시지에 충분한 정보를 담아 Consumer가 DB 재조회 없이 독립 처리할 수 있게 하고, 멱등성을 보장해서 중복 처리에 안전하게 만든다

관련 문서