메시지 전달의 각 단계별 신뢰성 보장 전략

RabbitMQ를 활용한 메시지 전달에서 신뢰성을 확보하려면 생산자에서 소비자까지 모든 단계가 안전하게 처리되어야 합니다. 하지만 신뢰성과 성능은 트레이드오프 관계에 있으므로, 비즈니스 요구사항에 따라 적절한 수준을 선택해야 합니다.

메시지 전달 경로는 다음과 같습니다:

1. 생산자 → RabbitMQ 브로커 → 교환기(Exchange) → 큐(Queue) → 소비자

생산자에서 교환기까지의 전달 성공 여부는 confirmCallback으로 확인하고, 교환기에서 큐까지의 라우팅 실패는 returnCallback으로 감지합니다.

생산자 → 교환기 구간 신뢰성

Confirm 모드

생산자가 메시지를 전송하면 교환기가 이를 수신했는지 확인 응답을 보내는 방식입니다. 이를 통해 메시지가 브로커에 정상적으로 도착했는지 확인할 수 있습니다.

설정 방법

• application.yml에서 확인 모드 활성화: spring.rabbitmq.publisher-confirm-type=correlated
• RabbitTemplate.ConfirmCallback 인터페이스 구현
• rabbitTemplate.setConfirmCallback()으로 콜백 등록

@Component
public class PublisherConfirmHandler implements RabbitTemplate.ConfirmCallback {

    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        log.info("Correlation ID: {}, ACK: {}, Cause: {}", 
            correlationData != null ? correlationData.getId() : null, ack, cause);

        if (!ack) {
            // 실패 시 재전송 로직 또는 로깅
            handleFailedDelivery(correlationData, cause);
        }
    }

    private void handleFailedDelivery(CorrelationData data, String cause) {
        // 실제 운영 환경에서는 재시도 큐에 추가하거나 알림 발송
    }
}

@Service
public class MessagePublisher {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private PublisherConfirmHandler confirmHandler;

    @PostConstruct
    public void initialize() {
        rabbitTemplate.setConfirmCallback(confirmHandler);
    }

    public void sendOrderMessage(Order order) {
        CorrelationData correlation = new CorrelationData(order.getOrderId());
        rabbitTemplate.convertAndSend("order.exchange", "order.create", order, correlation);
    }
}

트랜잭션 모드

RabbitMQ는 트랜잭션을 통해 메시지 전달을 보장합니다. 주요 메서드는 다음과 같습니다:

• txSelect(): 채널을 트랜잭션 모드로 전환
• txCommit(): 트랜잭션 커밋
• txRollback(): 트랜잭션 롤백

Channel channel = connection.createChannel();
try {
    channel.txSelect();
    channel.basicPublish("exchange.name", "routing.key", 
        MessageProperties.PERSISTENT_TEXT_PLAIN, message.getBytes());
    channel.txCommit();
} catch (Exception e) {
    channel.txRollback();
    // 실패 처리 로직
}

주의사항: 트랜잭션 모드는 성능 저하가 크고 동기 방식으로 동작하므로, Confirm 모드가 더 널리 사용됩니다.

교환기 → 큐 구간 신뢰성

라우팅 키 오류나 큐가 존재하지 않는 경우 메시지가 전달되지 않을 수 있습니다. 이를 처리하는 두 가지 방법이 있습니다.

Return 모드

설정 파일에서 반환 모드를 활성화합니다:

spring.rabbitmq.publisher-returns: true

@Component
public class MessageReturnHandler implements RabbitTemplate.ReturnsCallback {

    @Override
    public void returnedMessage(ReturnedMessage returned) {
        log.warn("메시지 반환됨 - Exchange: {}, RoutingKey: {}, Body: {}", 
            returned.getExchange(), returned.getRoutingKey(), 
            new String(returned.getMessage().getBody()));

        // 실패한 메시지에 대한 처리 (로깅, 재시도 등)
    }
}

@Service
public class MessagePublisher {

    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private MessageReturnHandler returnHandler;

    @PostConstruct
    public void initialize() {
        rabbitTemplate.setReturnsCallback(returnHandler);
    }

    public void sendMessage() {
        rabbitTemplate.convertAndSend("main.exchange", "invalid.key", "test message");
    }
}

백업 교환기(Alternate Exchange)

라우팅할 수 없는 메시지를 백업 교환기로 전달하는 방법입니다. 일반적으로 Fanout 타입으로 설정하여 모든 메시지를 수집합니다.

@Configuration
public class RabbitConfig {

    public static final String MAIN_EXCHANGE = "main.exchange";
    public static final String ALTERNATE_EXCHANGE = "alternate.exchange";
    public static final String DLQ_QUEUE = "dead.letter.queue";

    @Bean
    public DirectExchange mainExchange() {
        Map<String, Object> args = new HashMap<>();
        args.put("alternate-exchange", ALTERNATE_EXCHANGE);
        return new DirectExchange(MAIN_EXCHANGE, true, false, args);
    }

    @Bean
    public FanoutExchange alternateExchange() {
        return new FanoutExchange(ALTERNATE_EXCHANGE, true, false);
    }

    @Bean
    public Queue deadLetterQueue() {
        return QueueBuilder.durable(DLQ_QUEUE).build();
    }

    @Bean
    public Binding alternateBinding() {
        return BindingBuilder.bind(deadLetterQueue()).to(alternateExchange());
    }
}

큐에서 메시지 저장 신뢰성

시스템 장애나 재시작 시 메시지 손실을 방지하기 위해 지속성 설정이 필요합니다.

큐 지속성

@Bean
public Queue persistentQueue() {
    return QueueBuilder.durable("important.queue").build();
}

교환기 지속성

@Bean
public DirectExchange persistentExchange() {
    return ExchangeBuilder.directExchange("persistent.exchange")
        .durable(true)
        .build();
}

메시지 지속성

MessageProperties props = new MessageProperties();
props.setDeliveryMode(MessageDeliveryMode.PERSISTENT);
Message message = new Message("content".getBytes(), props);

클러스터와 미러 큐

고가용성을 위해 미러 큐(Mirror Queue)를 도입하면 단일 노드 장애 시에도 서비스가 지속됩니다. 미러 큐는 큐의 내용을 클러스터 내 여러 노드에 복제하여, 한 노드가 다운되어도 다른 노드가 작업을 이어받을 수 있습니다.

큐 → 소비자 구간 신뢰성

소비자가 메시지를 처리한 후에만 큐에서 제거되도록 수동 확인(Manual ACK) 방식을 사용합니다.

수동 ACK 설정

spring.rabbitmq.listener.simple.acknowledge-mode=manual

@RabbitListener(queues = "process.queue")
public void handleMessage(String message, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
    try {
        // 비즈니스 로직 처리
        processData(message);
        channel.basicAck(tag, false);
    } catch (Exception e) {
        log.error("메시지 처리 실패: {}", message, e);
        try {
            // requeue를 false로 설정하여 데드 레터 큐로 이동
            channel.basicNack(tag, false, false);
        } catch (IOException ex) {
            log.error("NACK 처리 실패", ex);
        }
    }
}

실패한 메시지는 basicNack()이나 basicReject()로 거부할 수 있으며, requeue 파라미터를 true로 설정하면 다시 큐에 추가됩니다. 무한 재시도를 방지하려면 별도의 데드 레터 큐를 구성하는 것이 좋습니다.