RabbitMQ 실무 활용 완벽 가이드

RabbitMQ를 실무에서 효과적으로 활용하는 방법을 알아봅니다. 메시지 큐의 기본 개념부터 고급 패턴, 성능 최적화, 장애 대응까지 실전에서 바로 쓸 수 있는 팁들을 담았습니다.

---

목차

1. Exchange 타입 선택 - 메시지 라우팅 전략의 핵심
2. 메시지 안정성 보장 - ACK와 Confirm 패턴
3. Dead Letter Exchange - 실패 메시지 처리 전략
4. Priority Queue - 중요한 메시지 먼저 처리
5. Consumer Prefetch - 부하 분산 최적화
6. Message TTL과 Queue TTL - 메시지 수명 관리
7. Lazy Queue - 대용량 메시지 처리
8. Connection과 Channel 관리 - 리소스 효율화

---

1. Exchange 타입 선택 - 메시지 라우팅 전략의 핵심

시작하며

여러분이 주문 서비스를 개발하는데, 주문이 들어오면 결제팀, 배송팀, 알림팀 모두에게 동시에 알려야 하는 상황을 겪어본 적 있나요? 각 팀에 일일이 API를 호출하다 보면 하나라도 실패하면 전체 프로세스가 막히고, 새로운 팀이 추가될 때마다 코드를 수정해야 하는 번거로움이 있습니다.

이런 문제는 실제 마이크로서비스 아키텍처에서 자주 발생합니다. 서비스 간 강한 결합으로 인해 확장성이 떨어지고, 한 서비스의 장애가 다른 서비스에 영향을 미치게 됩니다.

바로 이럴 때 필요한 것이 RabbitMQ의 Exchange입니다. Exchange 타입을 적절히 선택하면 메시지를 어떻게 라우팅할지 유연하게 제어할 수 있고, 서비스 간 결합도를 낮출 수 있습니다.

개요

간단히 말해서, Exchange는 RabbitMQ에서 메시지를 받아서 적절한 큐로 라우팅하는 메시지 중개자입니다. Exchange는 총 4가지 타입(Direct, Fanout, Topic, Headers)이 있으며, 각각 다른 라우팅 전략을 제공합니다.

예를 들어, 주문 이벤트를 여러 서비스에 동시에 브로드캐스트해야 한다면 Fanout을, 로그 레벨에 따라 선택적으로 처리해야 한다면 Topic을 사용하면 매우 유용합니다. 전통적으로는 각 서비스에 직접 HTTP 요청을 보내거나 DB 폴링을 사용했다면, Exchange를 사용하면 메시지를 한 번만 발행하고 라우팅 규칙에 따라 자동으로 분배할 수 있습니다.

Direct는 정확한 라우팅 키 매칭, Fanout은 모든 큐에 브로드캐스트, Topic은 패턴 매칭, Headers는 메시지 헤더 기반 라우팅을 제공합니다. 이러한 특징들이 서비스 간 느슨한 결합과 유연한 메시지 흐름 제어를 가능하게 합니다.

코드 예제

import pika

# RabbitMQ 연결 설정
connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Topic Exchange 선언 - 패턴 기반 라우팅
channel.exchange_declare(exchange='orders', exchange_type='topic', durable=True)

# 큐 생성 및 바인딩
channel.queue_declare(queue='payment_queue', durable=True)
channel.queue_declare(queue='shipping_queue', durable=True)
channel.queue_declare(queue='notification_queue', durable=True)

# 라우팅 키 패턴으로 바인딩
channel.queue_bind(exchange='orders', queue='payment_queue', routing_key='order.payment.*')
channel.queue_bind(exchange='orders', queue='shipping_queue', routing_key='order.shipping.*')
channel.queue_bind(exchange='orders', queue='notification_queue', routing_key='order.*')

# 메시지 발행 - 라우팅 키에 따라 자동 라우팅됨
channel.basic_publish(exchange='orders', routing_key='order.payment.created', body='{"order_id": 12345, "amount": 50000}')

설명

이것이 하는 일: Exchange는 Producer로부터 메시지를 받아서 바인딩 규칙에 따라 하나 이상의 큐로 메시지를 복사하여 전달합니다. 첫 번째로, exchange_declare로 Exchange를 생성합니다.

exchange_type='topic'은 와일드카드 패턴 매칭을 지원하는 타입으로, 'order.*'처럼 유연한 라우팅 규칙을 만들 수 있습니다. durable=True는 RabbitMQ 재시작 후에도 Exchange가 유지되도록 합니다.

그 다음으로, queue_declare로 각 서비스용 큐를 생성하고 queue_bind로 Exchange와 큐를 연결합니다. routing_key 패턴에서 '*'는 정확히 한 단어를 매칭하고, '#'는 0개 이상의 단어를 매칭합니다.

'order.payment.*'는 'order.payment.created', 'order.payment.confirmed' 같은 키와 매칭됩니다. 마지막으로, basic_publish로 메시지를 발행할 때 routing_key를 지정하면, Exchange가 바인딩 패턴과 비교하여 매칭되는 모든 큐에 메시지를 전달합니다.

'order.payment.created'는 payment_queue와 notification_queue 둘 다 매칭되므로 두 큐 모두에 전달됩니다. 여러분이 이 코드를 사용하면 새로운 서비스 추가 시 코드 수정 없이 큐 바인딩만 추가하면 되고, 각 서비스는 독립적으로 메시지를 처리할 수 있으며, 한 서비스의 장애가 다른 서비스에 영향을 주지 않습니다.

Topic Exchange를 사용하면 로그 레벨별 필터링, 지역별 라우팅, 우선순위별 처리 등 복잡한 비즈니스 로직을 선언적으로 구현할 수 있습니다.

실전 팁

💡 Fanout은 가장 빠르지만 선택적 라우팅이 불가능하므로, 모든 구독자에게 동일한 메시지를 보낼 때만 사용하세요. 조건부 라우팅이 필요하면 Topic을 선택하세요.

💡 Direct Exchange 사용 시 라우팅 키를 서비스 이름 대신 메시지 타입으로 설정하면 나중에 서비스 구조 변경 시 라우팅 로직을 수정하지 않아도 됩니다.

💡 Topic Exchange에서 '#'를 과도하게 사용하면 의도치 않은 큐에 메시지가 전달될 수 있으니, 가능한 구체적인 패턴을 사용하고 바인딩 규칙을 문서화하세요.

💡 Exchange는 반드시 durable=True로 선언하여 브로커 재시작 시에도 유지되도록 하고, 큐와 바인딩도 동일하게 설정하세요.

💡 개발 초기에는 Direct로 시작하고, 라우팅 요구사항이 복잡해지면 Topic으로 전환하세요. 처음부터 과도하게 복잡한 패턴을 사용하면 디버깅이 어렵습니다.

---

2. 메시지 안정성 보장 - ACK와 Confirm 패턴

시작하며

여러분이 결제 시스템을 운영하는데, 메시지를 보냈다고 생각했는데 실제로는 전달되지 않았거나, Consumer가 처리 중에 크래시되어 메시지가 사라진 적 있나요? 금융이나 주문 시스템처럼 메시지 손실이 치명적인 서비스에서는 이런 상황이 큰 문제가 됩니다.

이런 문제는 네트워크 불안정, 서버 장애, 처리 중 예외 등 다양한 원인으로 발생합니다. 메시지가 손실되면 주문이 누락되거나 결제가 처리되지 않는 등 비즈니스에 직접적인 영향을 미칩니다.

바로 이럴 때 필요한 것이 ACK(Acknowledgement)와 Publisher Confirm입니다. 이 두 메커니즘을 함께 사용하면 Producer부터 Consumer까지 전 구간에서 메시지가 안전하게 전달되고 처리되는 것을 보장할 수 있습니다.

개요

간단히 말해서, ACK는 Consumer가 메시지를 성공적으로 처리했음을 RabbitMQ에 알리는 신호이고, Publisher Confirm은 Producer가 메시지가 RabbitMQ에 안전하게 저장되었음을 확인하는 메커니즘입니다. ACK를 사용하지 않으면 RabbitMQ는 메시지를 전달하자마자 큐에서 삭제하므로, Consumer가 처리 중 실패하면 메시지가 영구 손실됩니다.

예를 들어, 데이터베이스 저장 중 Connection timeout이 발생하거나 서버가 갑자기 재시작되는 경우에 메시지를 복구할 방법이 없습니다. 기존에는 auto_ack=True로 설정하여 자동으로 ACK를 보냈다면, 이제는 auto_ack=False로 수동 ACK를 사용하여 처리가 완전히 끝난 후에만 확인 신호를 보낼 수 있습니다.

Manual ACK는 처리 성공 시 basic_ack, 재처리 요청 시 basic_nack, 단일 메시지 거부 시 basic_reject를 제공하고, Publisher Confirm은 메시지가 디스크에 저장되거나 모든 큐에 복제된 후 확인을 보냅니다. 이러한 특징들이 At-Least-Once 전달 보장을 가능하게 합니다.

코드 예제

import pika
import json
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Publisher Confirm 활성화 - 메시지 전달 보장
channel.confirm_delivery()

# 메시지 발행 with 확인
try:
    channel.basic_publish(
        exchange='orders',
        routing_key='order.created',
        body=json.dumps({'order_id': 12345}),
        properties=pika.BasicProperties(delivery_mode=2),  # 메시지 영속성
        mandatory=True  # 라우팅 실패 시 예외 발생
    )
    print("메시지 발행 성공 및 저장 확인됨")
except pika.exceptions.UnroutableError:
    print("라우팅 실패: 해당하는 큐가 없음")
except pika.exceptions.NackError:
    print("RabbitMQ가 메시지 저장 실패")

# Consumer with Manual ACK
def callback(ch, method, properties, body):
    try:
        # 메시지 처리 로직
        order = json.loads(body)
        process_order(order)  # 실제 비즈니스 로직

        # 처리 성공 시에만 ACK
        ch.basic_ack(delivery_tag=method.delivery_tag)
        print(f"주문 {order['order_id']} 처리 완료")
    except Exception as e:
        print(f"처리 실패: {e}")
        # 재처리를 위해 큐에 다시 넣기
        ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True)

channel.basic_qos(prefetch_count=1)  # 한 번에 하나씩 처리
channel.basic_consume(queue='order_queue', on_message_callback=callback, auto_ack=False)

설명

이것이 하는 일: Publisher Confirm과 Manual ACK를 조합하여 메시지가 Producer → RabbitMQ → Consumer 전체 경로에서 손실 없이 전달되고 처리되는 것을 보장합니다. 첫 번째로, Producer 측에서 channel.confirm_delivery()로 Confirm 모드를 활성화합니다.

이제 basic_publish는 동기 방식으로 동작하여 메시지가 RabbitMQ에 저장될 때까지 블로킹됩니다. delivery_mode=2는 메시지를 디스크에 영속화하고, mandatory=True는 라우팅 가능한 큐가 없으면 UnroutableError를 발생시킵니다.

그 다음으로, Consumer 측에서 auto_ack=False로 설정하여 수동 ACK 모드를 활성화합니다. callback 함수 내에서 실제 비즈니스 로직(process_order)이 성공적으로 완료된 후에만 basic_ack를 호출합니다.

만약 예외가 발생하면 basic_nack(requeue=True)로 메시지를 큐에 다시 넣어 재처리를 시도합니다. 마지막으로, basic_qos(prefetch_count=1)로 Consumer가 한 번에 하나의 메시지만 받도록 제한합니다.

이렇게 하면 처리가 느린 Consumer에게 메시지가 몰리지 않고, 여러 Consumer가 있을 때 부하가 고르게 분산됩니다. ACK를 보내기 전까지 RabbitMQ는 해당 메시지를 큐에 유지하므로, Consumer가 크래시되어도 다른 Consumer가 메시지를 받아 처리할 수 있습니다.

여러분이 이 코드를 사용하면 네트워크 장애나 서버 크래시 상황에서도 메시지 손실을 방지할 수 있고, 처리 실패 시 자동 재시도가 가능하며, 금융이나 주문 같은 크리티컬한 시스템에서 요구하는 메시지 안정성을 확보할 수 있습니다. 다만 동기 방식이므로 처리량은 약간 감소하지만, 안정성과 성능의 트레이드오프를 고려하여 선택해야 합니다.

실전 팁

💡 Publisher Confirm은 성능 오버헤드가 있으므로, 로그나 통계처럼 일부 손실이 허용되는 메시지는 Confirm 없이 발행하고, 주문이나 결제처럼 중요한 메시지만 Confirm을 사용하세요.

💡 basic_nack(requeue=True) 사용 시 무한 재시도를 방지하려면 메시지 헤더에 재시도 횟수를 기록하고, 일정 횟수 초과 시 Dead Letter Queue로 보내세요.

💡 prefetch_count는 Consumer의 처리 속도와 메모리를 고려하여 설정하세요. 너무 크면 메시지가 Consumer 메모리에 쌓이고, 너무 작으면 네트워크 왕복이 많아져 처리량이 감소합니다. 보통 10-50 사이가 적절합니다.

💡 delivery_mode=2는 디스크 I/O를 발생시키므로 성능이 중요하면 메시지를 배치로 모아서 발행하거나, Lazy Queue를 사용하여 디스크 쓰기를 최적화하세요.

💡 장시간 처리가 필요한 작업은 Consumer timeout 설정을 늘리거나, 주기적으로 heartbeat를 보내 RabbitMQ가 연결을 끊지 않도록 하세요.

---

3. Dead Letter Exchange - 실패 메시지 처리 전략

시작하며

여러분이 이메일 발송 서비스를 운영하는데, 일부 메시지가 계속 실패하여 큐에서 반복적으로 재처리되면서 시스템 전체 성능이 저하되는 상황을 겪어본 적 있나요? 잘못된 데이터나 외부 API 장애로 인해 절대 성공할 수 없는 메시지가 계속 재시도되면 정상 메시지 처리까지 지연됩니다.

이런 문제는 예외 처리 로직이 없거나, 재시도 횟수 제한이 없을 때 발생합니다. 실패한 메시지가 큐를 막아버리고, 개발자는 어떤 메시지가 왜 실패했는지 추적하기도 어렵습니다.

바로 이럴 때 필요한 것이 Dead Letter Exchange(DLX)입니다. DLX를 설정하면 처리 실패한 메시지를 자동으로 별도 큐로 이동시켜, 정상 메시지 처리는 계속하면서 실패 원인을 분석하고 재처리 전략을 수립할 수 있습니다.

개요

간단히 말해서, Dead Letter Exchange는 처리 실패, TTL 초과, 큐 길이 초과 등의 이유로 정상 큐에서 처리할 수 없는 메시지를 받아 처리하는 특별한 Exchange입니다. DLX는 원본 큐에서 메시지가 reject/nack(requeue=false)되거나, TTL이 만료되거나, 큐 길이가 max-length를 초과할 때 자동으로 작동합니다.

예를 들어, 외부 API 호출이 3번 실패한 메시지를 DLQ(Dead Letter Queue)로 보내고, 나중에 배치로 재처리하거나 수동으로 분석할 수 있습니다. 전통적으로는 try-catch로 예외를 잡아 별도 테이블에 저장했다면, DLX를 사용하면 RabbitMQ가 자동으로 실패 메시지를 분류하고 라우팅합니다.

DLX는 원본 메시지의 모든 정보를 보존하고, 실패 이유와 횟수를 헤더에 추가하며, 실패 메시지를 위한 별도의 처리 로직을 구현할 수 있게 합니다. 이러한 특징들이 장애 격리와 복구 전략 수립을 가능하게 합니다.

코드 예제

import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Dead Letter Exchange 선언
channel.exchange_declare(exchange='dlx', exchange_type='direct', durable=True)

# Dead Letter Queue 생성
channel.queue_declare(queue='dead_letter_queue', durable=True)
channel.queue_bind(exchange='dlx', queue='dead_letter_queue', routing_key='failed')

# 메인 큐 생성 with DLX 설정
channel.queue_declare(
    queue='email_queue',
    durable=True,
    arguments={
        'x-dead-letter-exchange': 'dlx',  # 실패 시 보낼 Exchange
        'x-dead-letter-routing-key': 'failed',  # DLX에서 사용할 라우팅 키
        'x-message-ttl': 3600000,  # 1시간 후 자동 DLQ 이동
    }
)

# Consumer with 재시도 횟수 추적
def callback(ch, method, properties, body):
    # 재시도 횟수 확인
    headers = properties.headers or {}
    retry_count = headers.get('x-retry-count', 0)

    try:
        message = json.loads(body)
        send_email(message)  # 이메일 발송 로직
        ch.basic_ack(delivery_tag=method.delivery_tag)
    except Exception as e:
        print(f"이메일 발송 실패 (시도 {retry_count + 1}/3): {e}")

        if retry_count < 2:  # 최대 3번 재시도
            # 재시도 횟수 증가 후 재발행
            new_headers = headers.copy()
            new_headers['x-retry-count'] = retry_count + 1

            ch.basic_publish(
                exchange='',
                routing_key='email_queue',
                body=body,
                properties=pika.BasicProperties(headers=new_headers, delivery_mode=2)
            )
            ch.basic_ack(delivery_tag=method.delivery_tag)
        else:
            # 최종 실패 - DLQ로 이동
            ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False)

channel.basic_consume(queue='email_queue', on_message_callback=callback, auto_ack=False)

설명

이것이 하는 일: Dead Letter Exchange를 사용하여 재시도 횟수를 제한하고, 최종 실패한 메시지를 자동으로 분리하여 정상 메시지 처리 흐름을 보호합니다. 첫 번째로, DLX와 DLQ를 미리 선언합니다.

'dlx'는 일반 Exchange와 동일하게 동작하며, 실패 메시지를 받아 'dead_letter_queue'로 라우팅합니다. 이 큐에는 나중에 별도의 Consumer를 붙여 실패 원인을 분석하거나 재처리할 수 있습니다.

그 다음으로, 메인 큐인 'email_queue'를 선언할 때 arguments로 DLX 설정을 추가합니다. 'x-dead-letter-exchange'는 메시지가 reject되거나 TTL이 초과될 때 보낼 Exchange를, 'x-dead-letter-routing-key'는 DLX에서 사용할 라우팅 키를 지정합니다.

'x-message-ttl'은 메시지가 큐에서 1시간 이상 처리되지 않으면 자동으로 DLQ로 이동시킵니다. 마지막으로, Consumer는 메시지 헤더에 재시도 횟수를 기록합니다.

처리 실패 시 retry_count가 2 미만이면 헤더를 업데이트하여 동일 큐에 재발행하고, 3번째 실패 시에는 basic_nack(requeue=False)로 거부하여 DLX로 자동 이동시킵니다. RabbitMQ는 DLQ로 이동된 메시지 헤더에 'x-death' 정보를 추가하여 실패 시각, 큐 이름, 이유 등을 기록합니다.

여러분이 이 코드를 사용하면 독성 메시지(poison message)로 인한 시스템 전체 마비를 방지할 수 있고, 실패 메시지를 모아서 배치로 분석하거나 재처리할 수 있으며, 알람 시스템과 연동하여 특정 타입의 실패가 증가하면 즉시 통보받을 수 있습니다. DLQ를 모니터링하면 외부 API 장애, 데이터 품질 문제, 버그 등을 조기에 발견할 수 있어 시스템 안정성이 크게 향상됩니다.

실전 팁

💡 DLQ도 반드시 Consumer를 붙여 주기적으로 모니터링하세요. DLQ가 계속 쌓이면 알람을 보내도록 설정하여 문제를 조기에 발견하세요.

💡 재시도 간격을 늘리려면 여러 단계의 큐를 만들고(retry_1m, retry_5m, retry_30m) 각각 다른 TTL을 설정하여 exponential backoff를 구현할 수 있습니다.

💡 DLQ에서 메시지를 다시 원본 큐로 보낼 때는 재시도 카운터를 반드시 초기화하고, 실패 원인이 해결되었는지 확인 후 재발행하세요.

💡 'x-message-ttl'은 큐 레벨로 설정하면 모든 메시지에 적용되고, 메시지별로 다르게 설정하려면 발행 시 properties.expiration을 사용하세요.

💡 DLX는 Consumer rejection뿐만 아니라 TTL 초과, 큐 길이 초과에도 작동하므로, 메모리 보호를 위해 max-length와 함께 사용하면 효과적입니다.

---

4. Priority Queue - 중요한 메시지 먼저 처리

시작하며

여러분이 고객 지원 티켓 시스템을 운영하는데, VIP 고객의 긴급 요청이 일반 문의들 뒤에 밀려서 몇 시간씩 대기하는 상황을 겪어본 적 있나요? 모든 메시지를 FIFO(First-In-First-Out) 순서로 처리하다 보면 중요도가 높은 작업이 적시에 처리되지 못해 고객 만족도가 떨어집니다.

이런 문제는 비즈니스 우선순위를 고려하지 않은 단순 큐 구조에서 발생합니다. 주문 취소 요청, 보안 알림, VIP 고객 문의 등은 일반 메시지보다 빨리 처리되어야 하지만, 일반 큐에서는 구분할 방법이 없습니다.

바로 이럴 때 필요한 것이 Priority Queue입니다. 메시지에 우선순위를 부여하면 RabbitMQ가 자동으로 높은 우선순위 메시지를 먼저 Consumer에게 전달하여, 중요한 작업이 신속하게 처리되도록 할 수 있습니다.

개요

간단히 말해서, Priority Queue는 메시지에 0-255 사이의 우선순위를 지정하여, 큐에 여러 메시지가 대기 중일 때 높은 우선순위 메시지가 먼저 전달되도록 하는 특수 큐입니다. Priority Queue는 큐 선언 시 'x-max-priority' 인자로 최대 우선순위 값을 설정하고, 메시지 발행 시 properties.priority로 개별 메시지의 우선순위를 지정합니다.

예를 들어, 주문 취소는 priority=9, 일반 주문은 priority=5, 통계 업데이트는 priority=1로 설정하여 비즈니스 중요도에 따라 처리 순서를 제어할 수 있습니다. 전통적으로는 여러 개의 별도 큐를 만들어 중요도별로 분리했다면, Priority Queue를 사용하면 하나의 큐에서 우선순위 기반 처리가 가능합니다.

Priority Queue는 Consumer가 유휴 상태일 때는 일반 큐처럼 즉시 전달하고, 메시지가 쌓여있을 때만 우선순위 정렬을 수행하며, 동일 우선순위는 FIFO 순서를 유지합니다. 이러한 특징들이 SLA가 다른 메시지들을 효율적으로 처리하게 합니다.

코드 예제

import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Priority Queue 선언 - 최대 우선순위 10
channel.queue_declare(
    queue='support_tickets',
    durable=True,
    arguments={'x-max-priority': 10}  # 0-10 범위의 우선순위 지원
)

# 다양한 우선순위로 메시지 발행
def create_ticket(customer_type, issue, priority):
    message = {
        'customer_type': customer_type,
        'issue': issue,
        'timestamp': time.time()
    }

    channel.basic_publish(
        exchange='',
        routing_key='support_tickets',
        body=json.dumps(message),
        properties=pika.BasicProperties(
            priority=priority,  # 우선순위 지정
            delivery_mode=2
        )
    )
    print(f"티켓 생성: {customer_type} (우선순위 {priority})")

# 실제 사용 예시
create_ticket('VIP', '결제 오류 - 긴급', priority=10)  # 최고 우선순위
create_ticket('Enterprise', '기능 문의', priority=7)
create_ticket('Standard', '일반 문의', priority=5)
create_ticket('Free', '피드백', priority=3)
create_ticket('System', '통계 업데이트', priority=1)  # 낮은 우선순위

# Consumer - 우선순위에 따라 자동으로 받음
def callback(ch, method, properties, body):
    ticket = json.loads(body)
    print(f"처리 중: {ticket['customer_type']} - {ticket['issue']} (우선순위 {properties.priority})")
    # 티켓 처리 로직
    process_ticket(ticket)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=1)
channel.basic_consume(queue='support_tickets', on_message_callback=callback, auto_ack=False)

설명

이것이 하는 일: Priority Queue를 사용하여 비즈니스 중요도에 따라 메시지 처리 순서를 제어하고, SLA가 다른 고객 유형을 차별화하여 서비스합니다. 첫 번째로, queue_declare에서 'x-max-priority' 인자로 큐가 지원할 최대 우선순위를 설정합니다.

여기서는 10으로 설정하여 0-10 범위의 우선순위를 사용할 수 있습니다. 이 값은 메모리 사용량에 영향을 미치므로 실제 필요한 만큼만 설정하세요(보통 3-10이 적절).

그 다음으로, basic_publish에서 properties.priority로 각 메시지의 우선순위를 지정합니다. VIP 고객의 긴급 이슈는 priority=10으로, 시스템 백그라운드 작업은 priority=1로 설정하여 비즈니스 규칙을 반영합니다.

우선순위를 지정하지 않으면 기본값 0이 사용됩니다. 마지막으로, Consumer는 특별한 설정 없이 basic_consume으로 메시지를 받습니다.

RabbitMQ가 내부적으로 우선순위를 고려하여 메시지를 전달하므로, Consumer는 자연스럽게 높은 우선순위 메시지를 먼저 받게 됩니다. 단, Consumer가 빠르게 처리하여 큐가 비어있으면 우선순위와 무관하게 도착 순서대로 즉시 전달됩니다.

여러분이 이 코드를 사용하면 고객 등급별 SLA를 구현할 수 있고, 시스템 과부하 시에도 중요한 작업이 우선 처리되며, 비즈니스 로직을 큐 레벨에서 선언적으로 표현할 수 있습니다. 예를 들어 주문 취소는 주문 생성보다 높은 우선순위로 설정하여, 고객이 빠르게 취소 확인을 받을 수 있도록 할 수 있습니다.

실전 팁

💡 x-max-priority는 너무 크게 설정하지 마세요. 우선순위 레벨이 많을수록 RabbitMQ가 내부적으로 관리할 구조가 복잡해져 메모리와 CPU를 더 사용합니다. 실무에서는 3-5 레벨이면 충분합니다.

💡 Priority Queue는 메시지가 쌓였을 때만 효과가 있으므로, Consumer 수를 적절히 조절하여 약간의 대기 시간이 생기도록 하세요. 큐가 항상 비어있으면 우선순위가 의미가 없습니다.

💡 모든 메시지에 동일한 높은 우선순위를 주면 Priority Queue의 이점이 사라지므로, 실제로 중요한 메시지에만 높은 우선순위를 부여하세요. 우선순위 인플레이션을 조심하세요.

💡 우선순위 규칙을 문서화하고 팀 전체가 공유하세요. 각 서비스가 제멋대로 우선순위를 설정하면 시스템 전체의 처리 순서가 예측 불가능해집니다.

💡 Priority Queue는 Lazy Queue와 함께 사용할 수 없으므로, 대용량 메시지를 처리하면서 우선순위도 필요하다면 메시지를 작게 나누거나 외부 스토리지를 활용하세요.

---

5. Consumer Prefetch - 부하 분산 최적화

시작하며

여러분이 여러 대의 Consumer 서버를 운영하는데, 어떤 서버는 너무 많은 메시지를 받아 처리가 느려지고, 어떤 서버는 대기 중인 상황을 겪어본 적 있나요? Consumer 성능이 제각각인데 모두에게 동일하게 메시지를 할당하면 느린 서버에 병목이 생기고 전체 처리량이 떨어집니다.

이런 문제는 RabbitMQ의 기본 Round-Robin 방식이 각 Consumer의 현재 부하 상태를 고려하지 않기 때문에 발생합니다. 특히 처리 시간이 들쑥날쑥한 작업이나, 서버 스펙이 다른 환경에서는 불균형이 심해집니다.

바로 이럴 때 필요한 것이 Prefetch Count입니다. Prefetch를 적절히 설정하면 각 Consumer가 처리 능력에 맞게 메시지를 가져가서, 빠른 Consumer는 더 많이, 느린 Consumer는 적게 처리하여 전체 처리량을 최대화할 수 있습니다.

개요

간단히 말해서, Prefetch Count는 RabbitMQ가 각 Consumer에게 한 번에 전달할 수 있는 최대 unacked 메시지 수를 제한하는 설정입니다. Prefetch Count를 설정하지 않으면(또는 0으로 설정) RabbitMQ는 큐의 모든 메시지를 Consumer들에게 균등하게 분배하므로, 느린 Consumer에게도 많은 메시지가 할당되어 메모리에 쌓이고 다른 Consumer는 놀게 됩니다.

예를 들어, 이미지 처리 작업처럼 시간이 오래 걸리는 경우 prefetch=1로 설정하면 처리 완료할 때마다 다음 메시지를 받아 부하가 자연스럽게 분산됩니다. 전통적으로는 별도의 로드 밸런서를 구현했다면, Prefetch를 사용하면 RabbitMQ가 자동으로 동적 부하 분산을 수행합니다.

Prefetch는 Consumer별, Channel별, Connection별로 설정할 수 있고, global=True로 설정하면 전체 Connection에 적용되며, Consumer가 ACK를 보낼 때마다 새 메시지를 받을 수 있습니다. 이러한 특징들이 Fair Dispatch(공정 분배)를 가능하게 합니다.

코드 예제

import pika
import time
import random

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

channel.queue_declare(queue='image_processing', durable=True)

# Prefetch 설정 - 핵심!
channel.basic_qos(
    prefetch_count=1,  # 한 번에 1개만 받음
    global_qos=False   # Channel별로 적용 (Connection 전체가 아님)
)

# 처리 시간이 다양한 작업 시뮬레이션
def callback(ch, method, properties, body):
    image_id = body.decode()

    # 이미지 크기에 따라 처리 시간이 다름 (1-10초)
    processing_time = random.randint(1, 10)
    print(f"이미지 {image_id} 처리 시작 (예상 {processing_time}초)")

    time.sleep(processing_time)  # 실제 이미지 처리 작업

    print(f"이미지 {image_id} 처리 완료")
    ch.basic_ack(delivery_tag=method.delivery_tag)
    # ACK 후 자동으로 다음 메시지를 받음

# 여러 Consumer가 실행 중이라고 가정
channel.basic_consume(
    queue='image_processing',
    on_message_callback=callback,
    auto_ack=False  # Manual ACK 필수
)

print("이미지 처리 Consumer 시작 (prefetch=1)")
channel.start_consuming()

# Prefetch 비교 예시
def optimized_consumer_example():
    """고성능 서버용 설정"""
    channel.basic_qos(prefetch_count=10)  # 10개씩 미리 가져옴
    # 빠른 처리가 가능하면 prefetch를 높여 네트워크 왕복 감소

def memory_constrained_example():
    """메모리가 부족한 환경"""
    channel.basic_qos(prefetch_count=1)  # 최소한으로 유지
    # 메시지가 크거나 메모리가 부족하면 prefetch=1

def balanced_example():
    """일반적인 웹 서비스"""
    channel.basic_qos(prefetch_count=20)  # 적당한 버퍼
    # 처리 속도와 메모리의 균형

설명

이것이 하는 일: basic_qos로 Prefetch Count를 설정하여 Consumer들이 처리 능력에 비례하여 메시지를 가져가도록 하고, Fair Dispatch를 통해 전체 처리량을 최대화합니다. 첫 번째로, basic_qos(prefetch_count=1)로 각 Consumer가 최대 1개의 unacked 메시지만 보유하도록 제한합니다.

이렇게 하면 Consumer가 메시지를 처리하고 ACK를 보내기 전까지 RabbitMQ는 추가 메시지를 보내지 않습니다. global_qos=False는 이 제한을 Channel별로 적용하므로, 하나의 Connection에 여러 Channel이 있어도 각각 독립적으로 동작합니다.

그 다음으로, callback 함수에서 실제 작업을 수행합니다. 이미지 처리처럼 시간이 오래 걸리는 작업의 경우 처리 완료 후 basic_ack를 호출하면, 그제야 RabbitMQ가 다음 메시지를 전달합니다.

만약 두 개의 Consumer가 있고 하나는 1초, 다른 하나는 10초가 걸린다면, 빠른 Consumer가 10개를 처리하는 동안 느린 Consumer는 1개만 처리하게 됩니다. 마지막으로, Prefetch 값은 서버 스펙과 작업 특성에 따라 조정해야 합니다.

CPU 바운드 작업이면 CPU 코어 수만큼, I/O 바운드 작업이면 더 높게(10-50), 메모리가 부족하거나 메시지가 크면 낮게(1-5) 설정합니다. 너무 낮으면 네트워크 왕복이 많아져 대기 시간이 늘어나고, 너무 높으면 메시지가 Consumer 메모리에 쌓여 다른 Consumer가 놀게 됩니다.

여러분이 이 코드를 사용하면 서버 스펙이 다른 이기종 환경에서도 효율적인 부하 분산이 가능하고, 느린 Consumer로 인한 전체 시스템 지연을 방지할 수 있으며, Auto Scaling 환경에서 새 Consumer가 추가되면 자동으로 부하가 분산됩니다. 또한 일부 Consumer가 장애로 느려지면 다른 Consumer가 자동으로 더 많은 작업을 처리하여 시스템 안정성이 향상됩니다.

실전 팁

💡 처리 시간이 예측 가능하고 균일하면 prefetch를 높여(20-100) 네트워크 오버헤드를 줄이고, 처리 시간이 들쑥날쑥하면 낮게(1-5) 설정하여 공정한 분배를 우선하세요.

💡 메시지 크기가 크면(수 MB 이상) prefetch를 낮게 유지하여 Consumer 메모리 사용량을 제한하세요. 특히 컨테이너 환경에서는 OOM Killer 방지가 중요합니다.

💡 Prefetch=0은 무제한을 의미하므로 절대 프로덕션에서 사용하지 마세요. 모든 메시지가 한 Consumer에 몰려 메모리가 부족해질 수 있습니다.

💡 모니터링 도구로 각 Consumer의 unacked 메시지 수를 추적하여, 특정 Consumer에 메시지가 계속 쌓이면 해당 서버의 문제를 조사하세요.

💡 Kubernetes 같은 오케스트레이션 환경에서는 Pod별 리소스 제한에 맞춰 prefetch를 조정하고, HPA(Horizontal Pod Autoscaler)와 함께 사용하면 트래픽에 따라 자동으로 처리 능력이 확장됩니다.

---

6. Message TTL과 Queue TTL - 메시지 수명 관리

시작하며

여러분이 실시간 알림 시스템을 운영하는데, 몇 시간 전의 오래된 알림이 지금 와서 전달되어 사용자에게 혼란을 주는 상황을 겪어본 적 있나요? "5분 전 주문이 도착했습니다"라는 알림이 2시간 뒤에 와서 이미 식은 음식을 받은 고객은 당황하게 됩니다.

이런 문제는 Consumer가 장시간 다운되거나, 큐에 메시지가 너무 많이 쌓여서 오래된 메시지가 그대로 유지될 때 발생합니다. 시간에 민감한 데이터는 유효기간이 지나면 의미가 없어지지만, 일반 큐는 이를 자동으로 처리하지 못합니다.

바로 이럴 때 필요한 것이 TTL(Time-To-Live)입니다. 메시지나 큐에 TTL을 설정하면 유효기간이 지난 데이터를 자동으로 제거하거나 Dead Letter Queue로 이동시켜, 항상 신선한 데이터만 처리할 수 있습니다.

개요

간단히 말해서, TTL은 메시지나 큐가 유지되는 최대 시간을 밀리초 단위로 지정하여, 지정된 시간이 지나면 자동으로 삭제되거나 Dead Letter Queue로 이동하는 메커니즘입니다. TTL은 두 가지 방식으로 설정할 수 있습니다: 1) 큐 레벨의 x-message-ttl은 해당 큐의 모든 메시지에 적용되고, 2) 메시지 레벨의 expiration은 개별 메시지마다 다른 TTL을 설정할 수 있습니다.

예를 들어, 실시간 주식 시세는 10초 TTL로, 배송 알림은 1시간 TTL로 설정하여 각 비즈니스 요구사항에 맞출 수 있습니다. 전통적으로는 Consumer에서 타임스탬프를 체크하여 오래된 메시지를 버렸다면, TTL을 사용하면 RabbitMQ가 자동으로 만료된 메시지를 제거하여 네트워크 트래픽과 처리 비용을 절약합니다.

메시지 TTL은 큐의 헤드에 있을 때만 체크되므로 즉시 삭제되지 않을 수 있고, Queue TTL은 큐가 일정 시간 사용되지 않으면 자동 삭제되며, DLX와 함께 사용하면 만료된 메시지를 별도 처리할 수 있습니다. 이러한 특징들이 리소스 효율성과 데이터 신선도를 보장합니다.

코드 예제

import pika
import json
import time

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# 방법 1: 큐 레벨 TTL - 모든 메시지에 동일 적용
channel.queue_declare(
    queue='realtime_notifications',
    durable=True,
    arguments={
        'x-message-ttl': 300000,  # 5분 (밀리초)
        'x-dead-letter-exchange': 'dlx',  # TTL 초과 시 DLX로 이동
        'x-dead-letter-routing-key': 'expired'
    }
)

# 방법 2: 메시지별 TTL - 개별 설정
def send_notification(user_id, message, ttl_seconds):
    """TTL이 다른 알림 발송"""
    channel.basic_publish(
        exchange='',
        routing_key='realtime_notifications',
        body=json.dumps({'user_id': user_id, 'message': message}),
        properties=pika.BasicProperties(
            expiration=str(ttl_seconds * 1000),  # 문자열로 밀리초 단위
            delivery_mode=2
        )
    )
    print(f"알림 발송 (TTL {ttl_seconds}초): {message}")

# 사용 예시
send_notification(123, "배달이 5분 내 도착합니다!", ttl_seconds=300)  # 5분
send_notification(456, "오늘의 할인 정보", ttl_seconds=3600)  # 1시간
send_notification(789, "실시간 주가 알림", ttl_seconds=10)  # 10초

# 방법 3: Queue TTL - 사용하지 않는 임시 큐 자동 삭제
channel.queue_declare(
    queue='temp_session_123',
    durable=False,
    arguments={
        'x-expires': 600000  # 10분간 Consumer 없으면 큐 자동 삭제
    }
)

# Consumer는 평소처럼 동작
def callback(ch, method, properties, body):
    notification = json.loads(body)
    # TTL 체크는 RabbitMQ가 자동으로 처리
    # 여기 도달한 메시지는 모두 유효기간 내
    print(f"알림 전달: {notification['message']}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_consume(queue='realtime_notifications', on_message_callback=callback, auto_ack=False)

설명

이것이 하는 일: TTL을 사용하여 시간에 민감한 메시지의 유효기간을 관리하고, 오래된 메시지로 인한 리소스 낭비와 잘못된 처리를 방지합니다. 첫 번째로, 큐 레벨 TTL은 queue_declare의 'x-message-ttl' 인자로 설정하며, 해당 큐의 모든 메시지에 동일한 TTL이 적용됩니다.

300000밀리초(5분) 후 메시지는 자동으로 삭제되거나, DLX가 설정되어 있으면 해당 Exchange로 이동합니다. 이 방식은 성능이 우수하지만 유연성은 떨어집니다.

그 다음으로, 메시지 레벨 TTL은 basic_publish의 properties.expiration으로 각 메시지마다 다른 유효기간을 설정할 수 있습니다. 주의할 점은 expiration은 반드시 문자열이어야 하고 밀리초 단위입니다.

이 방식은 유연하지만, RabbitMQ가 큐 헤드의 메시지 TTL만 체크하므로 중간에 있는 만료된 메시지는 즉시 삭제되지 않고 헤드에 도달할 때 삭제됩니다. 마지막으로, Queue TTL('x-expires')은 큐 자체의 수명을 관리합니다.

지정된 시간 동안 Consumer가 연결되지 않거나 메시지 consume이 없으면 큐가 자동으로 삭제됩니다. 이는 WebSocket 세션이나 임시 작업용 큐처럼 동적으로 생성되는 큐의 리소스 누수를 방지하는 데 유용합니다.

여러분이 이 코드를 사용하면 오래된 알림으로 인한 사용자 혼란을 방지할 수 있고, 메모리와 디스크 공간을 효율적으로 관리하며, DLX와 결합하여 만료된 메시지를 로깅하거나 분석할 수 있습니다. 예를 들어 만료 메시지가 급증하면 Consumer 성능 문제나 트래픽 폭증을 감지할 수 있습니다.

실전 팁

💡 메시지 레벨 TTL은 큐 헤드 메시지만 체크하므로, 짧은 TTL과 긴 TTL 메시지가 섞이면 긴 TTL 메시지가 짧은 TTL 메시지의 만료를 막을 수 있습니다. 이런 경우 큐 레벨 TTL을 사용하거나 큐를 분리하세요.

💡 TTL과 DLX를 함께 사용하면 만료된 메시지를 분석하여 시스템 병목을 찾을 수 있습니다. 특정 타입의 메시지 만료율이 높다면 Consumer를 추가하거나 TTL을 늘려야 합니다.

💡 실시간성이 중요한 알림은 TTL을 짧게(1-10분), 배치 처리 가능한 작업은 길게(수 시간~하루) 설정하여 비즈니스 요구사항에 맞추세요.

💡 Queue TTL('x-expires')은 큐가 비어있어도 Consumer가 연결되어 있으면 삭제되지 않습니다. 완전히 사용하지 않는 큐만 제거됩니다.

💡 TTL은 밀리초 단위이므로 계산 실수를 조심하세요. 5분은 300000(5 * 60 * 1000)이고, 하루는 86400000(24 * 60 * 60 * 1000)입니다. 상수로 정의하여 가독성을 높이세요.

---

7. Lazy Queue - 대용량 메시지 처리

시작하며

여러분이 수백만 건의 메시지를 처리하는 배치 작업을 운영하는데, RabbitMQ 메모리가 급격히 증가하여 서버가 느려지거나 메모리 부족으로 메시지 발행이 차단되는 상황을 겪어본 적 있나요? 대량의 메시지가 큐에 쌓이면 RabbitMQ는 기본적으로 모두 메모리에 유지하려고 하므로 수 GB의 RAM이 순식간에 소진됩니다.

이런 문제는 RabbitMQ의 기본 동작이 처리량을 위해 메시지를 메모리에 캐싱하는 방식이기 때문에 발생합니다. 큐가 길어질수록 메모리 사용량이 선형적으로 증가하여 결국 메모리 임계치에 도달하면 Publisher를 차단하거나 페이징으로 인해 성능이 급격히 저하됩니다.

바로 이럴 때 필요한 것이 Lazy Queue입니다. Lazy Queue는 메시지를 메모리 대신 디스크에 즉시 저장하여, 수백만 건의 메시지가 쌓여도 메모리 사용량이 안정적으로 유지되고 시스템이 안정적으로 동작합니다.

개요

간단히 말해서, Lazy Queue는 메시지를 메모리에 최소한만 유지하고 대부분 디스크에 저장하여, 큐 길이가 길어져도 메모리 사용량이 일정하게 유지되는 특수 큐 모드입니다. Lazy Queue는 큐 선언 시 'x-queue-mode': 'lazy' 인자로 활성화하며, 메시지가 도착하면 즉시 디스크에 쓰고 Consumer가 요청할 때만 디스크에서 읽어 전달합니다.

예를 들어, 야간 배치 작업으로 하루 동안 쌓인 수백만 건의 주문을 처리하거나, 트래픽 급증 시 메시지를 버퍼링할 때 매우 유용합니다. 전통적인 큐는 메시지를 메모리에 캐싱하다가 임계치를 넘으면 디스크로 페이징했다면, Lazy Queue는 처음부터 디스크 기반으로 동작하여 예측 가능한 성능을 제공합니다.

Lazy Queue는 메모리 사용량을 크게 줄이고(수십 MB 수준 유지), 수백만 건 메시지 처리가 가능하며, 큐 길이와 무관하게 안정적인 성능을 보이지만, 메시지 접근 시 디스크 I/O가 발생하여 지연시간은 약간 증가합니다. 이러한 특징들이 대용량 배치 처리와 안정적인 버퍼링을 가능하게 합니다.

코드 예제

import pika
import json

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Lazy Queue 선언 - 디스크 기반
channel.queue_declare(
    queue='batch_orders',
    durable=True,  # 재시작 후에도 유지
    arguments={
        'x-queue-mode': 'lazy'  # Lazy 모드 활성화
    }
)

# 대량 메시지 발행 - 메모리 걱정 없음
def publish_batch_orders(order_count):
    """수백만 건의 주문을 빠르게 발행"""
    print(f"{order_count}건의 주문 발행 시작...")

    for i in range(order_count):
        order = {
            'order_id': i,
            'customer_id': f'CUST_{i}',
            'amount': 50000,
            'created_at': time.time()
        }

        channel.basic_publish(
            exchange='',
            routing_key='batch_orders',
            body=json.dumps(order),
            properties=pika.BasicProperties(delivery_mode=2)
        )

        if i % 10000 == 0:
            print(f"{i}건 발행 완료... (메모리 사용량 안정)")

    print(f"총 {order_count}건 발행 완료!")

# 실제 사용 - 100만 건도 문제없음
publish_batch_orders(1000000)

# 일반 큐 vs Lazy Queue 비교
def compare_queue_modes():
    # 일반 큐 - 메시지 많으면 메모리 폭증
    channel.queue_declare(
        queue='normal_queue',
        durable=True
        # x-queue-mode 없음 = default mode
    )

    # Lazy Queue - 메시지 많아도 메모리 안정
    channel.queue_declare(
        queue='lazy_queue',
        durable=True,
        arguments={'x-queue-mode': 'lazy'}
    )

# Consumer는 동일하게 동작
def callback(ch, method, properties, body):
    order = json.loads(body)
    # 디스크에서 읽는 과정은 자동으로 처리됨
    process_order(order)
    ch.basic_ack(delivery_tag=method.delivery_tag)

channel.basic_qos(prefetch_count=100)  # Lazy Queue는 prefetch를 높여도 안전
channel.basic_consume(queue='batch_orders', on_message_callback=callback, auto_ack=False)

설명

이것이 하는 일: Lazy Queue를 사용하여 메모리 제약 없이 대용량 메시지를 처리하고, 트래픽 급증 시에도 시스템이 안정적으로 동작하도록 합니다. 첫 번째로, queue_declare에서 'x-queue-mode': 'lazy'로 Lazy 모드를 활성화합니다.

이제 이 큐로 발행되는 모든 메시지는 도착 즉시 디스크에 기록되고, 메모리에는 최소한의 메타데이터만 유지됩니다. durable=True와 함께 사용하면 RabbitMQ 재시작 후에도 메시지가 유지됩니다.

그 다음으로, 대량 메시지를 발행합니다. 일반 큐였다면 수만 건부터 메모리가 급증하기 시작하지만, Lazy Queue는 100만 건을 발행해도 메모리 사용량이 수십 MB 수준을 유지합니다.

디스크 I/O는 OS의 페이지 캐시와 RabbitMQ의 배치 쓰기로 최적화되므로 발행 성능도 양호합니다. 마지막으로, Consumer는 메시지를 요청하면 RabbitMQ가 디스크에서 읽어서 전달합니다.

이 과정은 자동으로 처리되므로 Consumer 코드는 일반 큐와 동일합니다. prefetch_count를 높여도(예: 100) 메모리에는 prefetch만큼만 로드되므로 안전합니다.

다만 디스크 읽기가 발생하므로 메시지당 지연시간은 일반 큐보다 약간 높지만(밀리초 단위), 처리량은 충분히 높습니다. 여러분이 이 코드를 사용하면 야간 배치 작업처럼 대량 메시지를 안전하게 버퍼링할 수 있고, 트래픽 급증 시 메모리 부족으로 시스템이 멈추는 것을 방지하며, 비용이 높은 RAM 대신 저렴한 디스크를 활용할 수 있습니다.

특히 클라우드 환경에서 메모리 비용을 크게 절감할 수 있습니다.

실전 팁

💡 실시간 처리가 중요하고 큐가 짧게 유지되는 경우는 일반 큐를, 대량 메시지를 버퍼링하거나 처리 속도보다 생성 속도가 빠른 경우는 Lazy Queue를 사용하세요.

💡 Lazy Queue는 디스크 기반이므로 SSD를 사용하면 성능이 크게 향상됩니다. HDD를 사용하면 IOPS 제약으로 처리량이 떨어질 수 있습니다.

💡 Lazy Queue는 Priority Queue와 호환되지 않습니다. 우선순위 처리가 필요하면 일반 큐를 사용하고 메모리를 충분히 할당하세요.

💡 모니터링에서 큐 길이(messages_ready)가 지속적으로 증가하면 Lazy Queue로 전환을 고려하세요. 일반 큐는 수만~수십만 건부터 메모리 문제가 발생합니다.

💡 기존 큐를 Lazy로 변경하려면 큐를 삭제하고 재생성해야 하므로, 처음부터 대용량 처리가 예상되면 Lazy로 만드는 것이 좋습니다. 큐 이름 규칙에 '_lazy'를 붙여 구분하세요.

---

8. Connection과 Channel 관리 - 리소스 효율화

시작하며

여러분이 멀티스레드 웹 애플리케이션을 개발하는데, 요청마다 RabbitMQ Connection을 새로 만들다가 "Too many connections" 오류가 발생하거나, Connection은 하나인데 여러 스레드가 동시에 사용하다가 "Channel closed unexpectedly" 오류를 겪어본 적 있나요? Connection과 Channel을 잘못 관리하면 리소스 낭비, 성능 저하, 예기치 않은 오류가 발생합니다.

이런 문제는 Connection과 Channel의 차이와 생명주기를 이해하지 못해서 발생합니다. Connection은 무겁고 비용이 높은 TCP 연결이고, Channel은 가벼운 가상 연결인데, 이를 혼동하면 비효율적인 아키텍처가 됩니다.

바로 이럴 때 필요한 것이 Connection Pool과 Thread-safe Channel 관리입니다. Connection은 애플리케이션당 소수만 유지하고, Channel은 스레드별로 생성하여 동시성 문제 없이 효율적으로 RabbitMQ를 사용할 수 있습니다.

개요

간단히 말해서, Connection은 RabbitMQ 서버와의 TCP 연결로 무겁고 비용이 높으며, Channel은 Connection 내부의 가상 연결로 가볍고 빠르게 생성할 수 있는 구조입니다. Connection 하나에 여러 개의 Channel을 만들 수 있으며, 각 Channel은 독립적으로 메시지를 송수신합니다.

예를 들어, 웹 서버에서 Connection은 애플리케이션 시작 시 1-2개만 생성하고, 각 스레드나 코루틴은 자신만의 Channel을 사용하여 동시성 문제를 피할 수 있습니다. 전통적으로는 매 요청마다 Connection을 생성하고 닫았다면, Connection Pool을 사용하면 연결을 재사용하여 오버헤드를 대폭 줄입니다.

Connection은 Heartbeat로 연결 상태를 유지하고, 네트워크 오류 시 재연결 로직이 필요하며, Channel은 Thread-safe하지 않으므로 스레드별로 생성해야 하고, 오류 발생 시 닫히므로 재생성해야 합니다. 이러한 특징들이 안정적이고 효율적인 RabbitMQ 사용을 가능하게 합니다.

코드 예제

import pika
import threading
from contextlib import contextmanager

class RabbitMQConnectionPool:
    """Thread-safe Connection Pool"""

    def __init__(self, host='localhost', max_connections=5):
        self.host = host
        self.max_connections = max_connections
        self.connections = []
        self.lock = threading.Lock()

    def get_connection(self):
        """Connection 가져오기 (재사용)"""
        with self.lock:
            if self.connections:
                return self.connections.pop()
            else:
                # 새 Connection 생성 (무거운 작업)
                return pika.BlockingConnection(
                    pika.ConnectionParameters(
                        host=self.host,
                        heartbeat=600,  # 10분마다 heartbeat
                        blocked_connection_timeout=300  # 5분 타임아웃
                    )
                )

    def return_connection(self, connection):
        """Connection 반환 (재사용을 위해)"""
        with self.lock:
            if len(self.connections) < self.max_connections:
                self.connections.append(connection)
            else:
                connection.close()

    @contextmanager
    def get_channel(self):
        """Channel Context Manager - 자동 정리"""
        connection = self.get_connection()
        channel = connection.channel()  # 가벼운 작업
        try:
            yield channel
        finally:
            channel.close()
            self.return_connection(connection)

# 전역 Connection Pool (애플리케이션 시작 시 한 번)
pool = RabbitMQConnectionPool(max_connections=5)

# Thread-safe 메시지 발행
def publish_message_safe(routing_key, message):
    """각 스레드가 자신만의 Channel 사용"""
    with pool.get_channel() as channel:
        channel.queue_declare(queue=routing_key, durable=True)
        channel.basic_publish(
            exchange='',
            routing_key=routing_key,
            body=message,
            properties=pika.BasicProperties(delivery_mode=2)
        )
        print(f"[Thread {threading.current_thread().name}] 메시지 발행 완료")

# 멀티스레드 환경에서 안전하게 사용
import concurrent.futures

def worker(task_id):
    """각 스레드가 독립적으로 메시지 발행"""
    publish_message_safe('tasks', f'Task {task_id}')

# 동시에 100개 스레드 실행
with concurrent.futures.ThreadPoolExecutor(max_workers=100) as executor:
    futures = [executor.submit(worker, i) for i in range(100)]
    concurrent.futures.wait(futures)

print("모든 작업 완료 - Connection은 최대 5개만 사용됨")

# 재연결 로직 예시
def create_connection_with_retry(max_retries=3):
    """네트워크 오류 시 자동 재연결"""
    for attempt in range(max_retries):
        try:
            return pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        except pika.exceptions.AMQPConnectionError as e:
            print(f"연결 실패 ({attempt + 1}/{max_retries}): {e}")
            time.sleep(2 ** attempt)  # Exponential backoff
    raise Exception("RabbitMQ 연결 실패")

설명

이것이 하는 일: Connection Pool을 구현하여 비용이 높은 Connection을 재사용하고, Thread-safe한 Channel 관리로 동시성 환경에서 안전하게 RabbitMQ를 사용합니다. 첫 번째로, RabbitMQConnectionPool 클래스는 제한된 수의 Connection을 유지하는 Pool을 구현합니다.

get_connection()은 Pool에서 Connection을 가져오거나 없으면 새로 생성하고, return_connection()은 사용 완료된 Connection을 Pool에 반환합니다. max_connections=5로 제한하여 RabbitMQ 서버에 과도한 연결이 생성되지 않도록 합니다.

그 다음으로, get_channel() Context Manager는 Connection에서 Channel을 생성하고, with 블록이 끝나면 자동으로 Channel을 닫고 Connection을 Pool에 반환합니다. Channel 생성/삭제는 매우 빠르므로(밀리초 미만) 매번 새로 만들어도 성능 문제가 없지만, Connection은 재사용하여 TCP handshake와 인증 오버헤드를 피합니다.

마지막으로, 멀티스레드 환경에서 각 스레드는 publish_message_safe를 통해 자신만의 Channel을 사용합니다. ThreadPoolExecutor로 100개 스레드가 동시 실행되어도, 실제 Connection은 Pool 크기인 5개만 사용되므로 리소스 효율적입니다.

Heartbeat=600은 10분마다 연결 유지 신호를 보내 방화벽이나 로드밸런서에서 연결이 끊기지 않도록 합니다. 여러분이 이 코드를 사용하면 웹 애플리케이션에서 요청마다 Connection을 생성하는 오버헤드를 제거할 수 있고, 멀티스레드나 비동기 환경에서 동시성 오류 없이 안전하게 메시지를 처리할 수 있으며, Connection 수 제한으로 RabbitMQ 서버 부하를 제어할 수 있습니다.

또한 재연결 로직을 추가하면 네트워크 불안정 상황에서도 자동 복구됩니다.

실전 팁

💡 Connection은 애플리케이션당 1-10개 정도면 충분하고, Channel은 스레드나 코루틴마다 생성하세요. Connection 수를 늘려도 성능이 향상되지 않으며 오히려 서버 부하만 증가합니다.

💡 Channel은 Thread-safe하지 않으므로 절대 여러 스레드에서 공유하지 마세요. 공유하면 "unexpected frame" 같은 프로토콜 오류가 발생합니다.

💡 Long-lived Consumer는 Connection을 계속 유지하고, Short-lived Publisher는 Connection Pool을 사용하는 것이 효율적입니다.

💡 Heartbeat 간격은 네트워크 환경에 따라 조정하세요. 방화벽이 60초 idle timeout을 가지면 heartbeat=30 정도로 설정하여 연결이 끊기지 않도록 하세요.

💡 pika의 BlockingConnection은 동기 방식이므로 비동기 프레임워크(asyncio, FastAPI)에서는 aio-pika 같은 비동기 라이브러리를 사용하세요. 아키텍처에 맞는 클라이언트를 선택하는 것이 중요합니다.

---