Message Queue

Message Queue(메시지 큐)는 생산자(Producer)와 소비자(Consumer) 사이에서 메시지를 임시 저장하고 전달하는 비동기 통신 시스템입니다. 동기식 API 호출과 달리, 생산자는 메시지를 큐에 넣고 즉시 반환되며, 소비자는 자신의 속도로 메시지를 처리합니다.

주요 장점: 1) 느슨한 결합 - 서비스 간 직접 의존 없이 메시지로 통신, 2) 비동기 처리 - 오래 걸리는 작업을 백그라운드로 분리, 3) 피크 부하 완화 - 트래픽 급증 시 큐가 버퍼 역할, 4) 신뢰성 - 메시지 영속화로 장애 시에도 데이터 손실 방지, 5) 확장성 - 컨슈머를 수평 확장하여 처리량 증가.

대표 솔루션: RabbitMQ(AMQP, 유연한 라우팅), Apache Kafka(고처리량 이벤트 스트리밍), AWS SQS(관리형, 간편), Redis Streams(경량, 빠름). 선택 기준은 처리량, 순서 보장, 메시지 크기, 영속성 요구사항에 따라 다릅니다.

일반적인 사용 사례: 이메일/알림 발송, 이미지/비디오 처리, 주문 처리 파이프라인, 로그 수집, 마이크로서비스 간 이벤트 전달. 동기식으로 처리하면 느려지거나 실패할 수 있는 작업을 큐로 분리합니다.

복사

// Message Queue 패턴 예제 (BullMQ - Redis 기반)
import { Queue, Worker, Job } from 'bullmq';
import Redis from 'ioredis';

const connection = new Redis(process.env.REDIS_URL);

// 1. 큐 생성
const emailQueue = new Queue('email', { connection });
const imageQueue = new Queue('image-processing', { connection });

// 2. 메시지(Job) 생산 - Producer
async function sendWelcomeEmail(userId: string, email: string) {
  // 즉시 반환, 이메일 발송은 비동기로
  await emailQueue.add('welcome', {
    userId,
    email,
    template: 'welcome'
  }, {
    attempts: 3,           // 실패 시 3번 재시도
    backoff: {
      type: 'exponential',
      delay: 1000          // 1초 → 2초 → 4초
    },
    removeOnComplete: 100, // 완료된 Job 100개까지만 유지
    removeOnFail: 1000
  });
}

// 이미지 처리 요청
async function processImage(imageId: string, operations: string[]) {
  await imageQueue.add('resize', {
    imageId,
    operations
  }, {
    priority: 1,  // 우선순위 (낮을수록 먼저)
    delay: 0      // 즉시 처리
  });
}

// 3. 메시지 소비 - Consumer (Worker)
const emailWorker = new Worker('email', async (job: Job) => {
  const { email, template, userId } = job.data;

  console.log(`Processing email job ${job.id}: ${email}`);

  // 실제 이메일 발송
  await sendEmail({
    to: email,
    template,
    data: { userId }
  });

  return { sent: true, email };  // 결과 저장
}, {
  connection,
  concurrency: 5,  // 동시 처리 수
  limiter: {
    max: 100,      // 분당 최대 100개
    duration: 60000
  }
});

// 이벤트 핸들링
emailWorker.on('completed', (job) => {
  console.log(`Job ${job.id} completed`);
});

emailWorker.on('failed', (job, err) => {
  console.error(`Job ${job?.id} failed:`, err);
  // 알림, 모니터링 등
});

// 4. AWS SQS 예제
import { SQSClient, SendMessageCommand, ReceiveMessageCommand } from '@aws-sdk/client-sqs';

const sqs = new SQSClient({ region: 'ap-northeast-2' });
const QUEUE_URL = process.env.SQS_QUEUE_URL;

// 메시지 전송
async function sendToSQS(data: object) {
  await sqs.send(new SendMessageCommand({
    QueueUrl: QUEUE_URL,
    MessageBody: JSON.stringify(data),
    MessageAttributes: {
      Type: {
        DataType: 'String',
        StringValue: 'order'
      }
    }
  }));
}

// 메시지 수신 (Long Polling)
async function pollMessages() {
  const response = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: QUEUE_URL,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,  // Long polling
    VisibilityTimeout: 30  // 처리 시간
  }));

  for (const message of response.Messages || []) {
    await processMessage(JSON.parse(message.Body!));
    // 처리 완료 후 삭제
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: QUEUE_URL,
      ReceiptHandle: message.ReceiptHandle
    }));
  }
}

RabbitMQ Kafka Event-Driven Pub/Sub Saga Pattern

📄 AWS - Message Queue 개요 📄 BullMQ - Node.js Queue 라이브러리 📄 RabbitMQ for Beginners