데드 레터 큐 핵심 개념

데드 레터 큐(Dead Letter Queue, DLQ)는 정상적으로 소비되지 못한 메시지를 처리하는 데 사용되며, 다음과 같은 시나리오에서 트리거됩니다:

1. 메시지가 거부(reject)되고 requeue=false 설정
2. 메시지 만료(TTL 초과)
3. 큐 최대 길이 도달

수학적으로 큐 상태는 다음과 같이 표현할 수 있습니다:

$$ \text{DLQ} = \{ m \in \text{Queue} \mid \text{TTL}(m) \leq 0 \lor \text{RejectCount}(m) \geq N \} $$

소비자 래핑 주요 설계

다음은 데드 레터 큐 시나리오를 위한 C++ 소비자 래퍼 클래스입니다.

#include <amqpcpp.h>
#include <functional>
#include <string>
#include <utility>

class ManagedDeadLetterConsumer {
private:
    AMQP::TcpChannel& m_channel;
    std::string m_processingQueue;
    std::string m_dlExchange;
    std::string m_dlRoutingKey;
    const int m_maxRetries;

    // 메시지 처리 콜백 함수 타입 정의
    using MessageProcessor = std::function<bool(const AMQP::Message&)>;

public:
    ManagedDeadLetterConsumer(AMQP::TcpChannel& channel,
                              std::string queueName,
                              int maxRetries = 3,
                              std::string dlExchange = "general.dlx",
                              std::string dlRoutingKey = "general.dl.key")
        : m_channel(channel),
          m_processingQueue(std::move(queueName)),
          m_maxRetries(maxRetries),
          m_dlExchange(std::move(dlExchange)),
          m_dlRoutingKey(std::move(dlRoutingKey))
    {
        // 데드 레터 정책 구성
        AMQP::Table queueArgs;
        queueArgs["x-dead-letter-exchange"] = m_dlExchange;
        queueArgs["x-dead-letter-routing-key"] = m_dlRoutingKey;
        m_channel.declareQueue(m_processingQueue, AMQP::durable, queueArgs);
    }

    // 소비자 시작 메서드
    void startConsuming(MessageProcessor processor) {
        m_channel.consume(m_processingQueue)
            .onReceived([this, processor](const AMQP::Message &message,
                                         uint64_t deliveryTag,
                                         bool redelivered) {
                try {
                    if (processor(message)) {
                        // 성공 시 메시지 승인
                        m_channel.ack(deliveryTag);
                    } else {
                        // 실패 시 메시지 재처리 시도 또는 DLQ로 라우팅
                        handleProcessingFailure(message, deliveryTag);
                    }
                } catch (const std::exception& ex) {
                    // 예외 발생 시 실패 처리
                    handleProcessingFailure(message, deliveryTag);
                    // 에러 로깅 또는 다른 처리
                }
            });
    }

private:
    // 메시지 처리 실패 시 로직
    void handleProcessingFailure(const AMQP::Message& msg, uint64_t deliveryTag) {
        AMQP::Table headers = msg.headers();
        int currentRetryCount = headers.contains("x-retry-count") ? headers.get("x-retry-count").asInt() : 0;

        if (currentRetryCount < m_maxRetries) {
            // 재시도 횟수 증가 및 원래 큐로 다시 발행
            headers["x-retry-count"] = currentRetryCount + 1;
            // 메시지를 원래 큐로 다시 보내 재시도
            m_channel.publish("", m_processingQueue, msg.body(), AMQP::mandatory, headers);
            // 원래 메시지는 승인하여 큐에서 제거
            m_channel.ack(deliveryTag);
        } else {
            // 최대 재시도 횟수 초과 시 거부 (DLQ로 이동)
            m_channel.reject(deliveryTag, false); // requeue = false
        }
    }
};
    

애플리케이션 시나리오 구현

다음은 위에서 정의한 ManagedDeadLetterConsumer를 사용하여 실제 애플리케이션을 구현하는 예제입니다.

// RabbitMQ 연결 및 채널 설정 (handler는 AMQP::LibEventReactor 또는 유사 객체)
AMQP::TcpConnection connection(&handler, "localhost", 5672);
AMQP::TcpChannel dataChannel(&connection);

// 데이터 처리 큐 이름과 데드 레터 관련 정보 설정
const std::string orderQueue = "order.processing.queue";
const std::string orderDlx = "order.dead.letter.exchange";
const std::string orderDlKey = "order.failure.key";
const int maxOrderRetries = 5;

// 데드 레터 기능을 갖춘 소비자 인스턴스 생성
ManagedDeadLetterConsumer orderConsumer(
    dataChannel,
    orderQueue,
    maxOrderRetries,
    orderDlx,
    orderDlKey
);

// 주문 데이터 처리 로직 정의
auto processOrderMessage = [](const AMQP::Message& msg) -> bool {
    try {
        // JSON 파싱 또는 기타 데이터 추출 로직
        // Order orderData = parseOrderFromJson(msg.body());
        // if (!orderData.isValid()) {
        //     // 유효성 검사 실패
        //     return false;
        // }
        // performOrderAction(orderData); // 실제 비즈니스 로직 수행
        // 성공적으로 처리되었음을 알림
        return true;
    } catch (const std::exception& e) {
        // 파싱 또는 처리 중 오류 발생
        // logError("Order processing error: " + std::string(e.what()));
        return false;
    }
};

// 소비자 시작
orderConsumer.startConsuming(processOrderMessage);

// 별도의 데드 레터 큐 소비자 설정 (선택 사항)
// ManagedDeadLetterConsumer dlqInspector(dataChannel, "order.dead.letter.exchange"); // DLX를 큐 이름으로 지정
// dlqInspector.startConsuming([](const AMQP::Message& msg) -> bool {
//     std::cerr << "Received dead letter: " << msg.body() << std::endl;
//     // 실제 로그 기록 또는 알림 시스템 연동
//     return true; // 데드 레터 소비 확인
// });
    

데드 레터 큐 처리 전략

1. 자동 재시도 메커니즘:

   메시지 처리 실패 시, 재시도 횟수를 헤더에 추가하여 원래 큐로 다시 발행합니다. 설정된 최대 재시도 횟수를 초과하면 거부하여 데드 레터 큐로 보냅니다.

   위의 handleProcessingFailure 메서드가 이 기능을 구현합니다.

2. 데드 레터 메시지 분석:

   별도의 소비자 인스턴스를 생성하여 데드 레터 큐(또는 DLX)에서 메시지를 수신하고, 해당 메시지에 대한 상세 정보를 기록하거나 알림을 보냅니다.

   애플리케이션 시나리오 구현 예제의 주석 처리된 부분을 참고하십시오.

성능 최적화 제안

1. QoS(Quality of Service) 제어:

   채널별 사전 가져오기(prefetch count) 수를 동적으로 조절하여, 동시에 처리 중인 메시지 수를 제한하고 과도한 메모리 사용을 방지합니다.

   $$ \text{PrefetchCount} = \text{min}(\text{ConsumerThreads} \times \text{Factor}, \text{MaxLimit}) $$

   
   // 예: prefetchCount = 10;
   // dataChannel.setQos(prefetchCount);
           

2. 연결 재사용:

   애플리케이션 내에서 RabbitMQ 연결 객체를 풀링(pooling)하여 관리하고, 필요한 채널만 동적으로 생성 및 해제함으로써 TCP 연결 설정 및 해제 오버헤드를 줄입니다.

3. 예외 격리:

   다양한 종류의 오류(예: 유효성 검사 오류, 결제 오류, 시스템 오류)에 대해 각각 다른 데드 레터 교환(exchange)이나 라우팅 키를 지정하여, 특정 유형의 오류가 다른 메시지 처리에 영향을 미치지 않도록 격리합니다.

   이를 위해 declareQueue 시 x-dead-letter-exchange 및 x-dead-letter-routing-key 인자를 다르게 설정할 수 있습니다.