RabbitMQ에 대해 알아보자 - (12) Producer 트랜잭션 전략

들어가기 앞서

이전 포스팅에서 비동기 메시징 시스템을 사용함에 있어, 메시지 전송의 신뢰성에 대해서 알아보았습니다.

RabbitMQ에 대해 알아보자 - (10) 트랜잭션 전략 / 신뢰성 있는 메시지

Producer에서 메시지를 발행했지만, RabbitMQ의 큐에 일시적인 장애로 인해 메시지를 수신하지 못 했다

 

이번 포스팅 에서는 Producer에서 신뢰성 있는 메시지를 전송하는 방법에 대해 알아볼 예정입니다. 특히 Producer 입장에서 메시지를 정확히 한 번 전송하고, 유실 없이 브로커에 도달시켜야 하는 요구사항이 있다면, 단순히 convertAndSend()를 호출하는 것만으로는 부족합니다.

 

RabbitMQ는 Producer가 메시지 전송의 신뢰성을 확보할 수 있도록 다음 두 가지 트랜잭션 방식을 제공합니다:

1. 트랜잭션 메시징 (Transaction Messaging)
2. Publisher Confirms

이 글에서는 각각의 특징과 동작 방식, 장단점 비교하면서 어떤 상황에서 어떤 전략을 선택하는 것이 좋은지를 정리해보겠습니다.

---

Producer 트랜잭션 전략

1. 트랜잭션 메시징 (Transaction Messaging)

트랜잭션 메시징은 RabbitMQ의 AMQP 프로토콜에서 제공하는 표준 트랜잭션 처리 방식입니다. 트랜잭션 메시징(Transaction Messaging)은 메시지 전송의 원자성(Atomicity) 을 보장하기 위해 AMQP의 트랜잭션 API를 활용하는 방식입니다. Producer는 txSelect(), txCommit(), txRollback() 명령을 통해 메시지를 큐에 커밋하거나 롤백할 수 있으며, 이를 통해 메시지가 브로커에 정확하게 도달했는지 여부를 직접 제어할 수 있습니다.

프로듀서는 메시지 전송 결과를 RabbitMQ로 부터 동기로 수신받는다

 

트랜잭션 메시징의 특징

1. 전송 원자성 보장
   트랜잭션 내에서 전송된 메시지는 모두 성공해야만 커밋됩니다. 하나라도 실패하면 전체 메시지 전송이 롤백되어 MQ에는 전혀 반영되지 않습니다.
2. 메시지 유실 방지
   메시지가 브로커에 안전하게 저장되었을 때만 커밋되므로, 전송 중 오류로 인한 유실을 방지할 수 있습니다.
3. 동기적 트랜잭션 보장
   txSelect()로 트랜잭션을 시작하고, txCommit()을 호출해서 커밋을 마무리하게 됩니다. 즉, 브로커로부터 커밋 결과가 돌아올 때까지 기다리는 구조로, 동기적인 처리 방식입니다.

 

트랜잭션 메시징의 한계

1. 성능 오버헤드
   트랜잭션은 디스크 동기화, 브로커와의 커밋/롤백 확인 등 부가적인 작업을 동반하기 때문에 메시지 처리 속도가 크게 저하됩니다. 일반적으로 TPS는 비트랜잭션 방식 대비 수십 분의 1 수준까지 떨어질 수 있습니다.
2. 구현 복잡성 증가
   트랜잭션 처리를 잘못 구성할 경우, 메시지 중복 전송 또는 손실이 발생할 수 있습니다. 트랜잭션 롤백 시점과 Ack/Nack 처리 흐름 간의 충돌은 안정적인 운영을 어렵게 만듭니다.
3. 분산 트랜잭션 미지원
   트랜잭션 메시징은 RabbitMQ 단일 브로커 내에서만 동작하며, DB 작업과 메시지 전송 간의 분산 트랜잭션은 보장하지 않습니다.
   메시지를 전송하기 전에 DB에 데이터를 저장하거나 반대로 메시지 전송 후 DB 작업을 수행해야 할 경우, 두 작업 사이의 원자성 확보가 불가능합니다. 이를 해결하기 위해선 다음과 같은 별도 메커니즘이 필요합니다.

---

2. Publisher Confirms

RabbitMQ 공식 문서에서도 트랜잭션 메시징보다는Publisher Confirms (AMQP Confirm Select) 방식을 추천합니다. Publisher Confirms는 RabbitMQ에서 제공하는 경량 메시지 전송 확인 메커니즘으로, 메시지를 브로커에 전송한 뒤 브로커로부터 ACK 또는 NACK 응답을 비동기적으로 수신하여 메시지의 수신 여부를 확인할 수 있는 방식입니다.

프로듀서는 메시지 전송 결과를 RabbitMQ로 부터 비동기로 수신 받는다

 

AMQP의 Confirm.Select 모드를 활용하는 기능으로, Producer가 메시지를 큐에 보낸 뒤에도 브로커에 안전하게 도달했는지 여부를 명확히 알 수 있게 해줍니다. 특히 성능이 중요한 시스템에서 트랜잭션 메시징의 대안으로 널리 사용됩니다.

 

Publisher Confirms의 특징

1. 가볍고 빠른 메시지 전송
   메시지를 전송한 후, 브로커로부터 메시지 수신 여부에 대한 ACK/NACK 응답을 비동기적으로 받습니다.
2. 비동기 + 비차단(Non-blocking) 구조
   메시지를 큐에 넣는 과정이 비동기이며, 전송 흐름이 차단되지 않아 처리량(TPS)이 높습니다.
3. 개별 메시지 재처리 가능
   트랜잭션처럼 전체 메시지를 롤백하지 않고, 전송에 실패한 개별 메시지만 선별적으로 재전송할 수 있습니다.
4. 응답 처리 방식 명확
   • ACK: 브로커가 메시지를 정상적으로 수신함
   • NACK: 브로커가 메시지를 수신하지 못했거나 처리에 실패함
5. 고성능 시스템에 적합
   동기 트랜잭션 방식보다 전송 성능이 2배 이상 높으며, Broker의 디스크 I/O 부담도 낮아 대량 메시지 처리 환경에서도 안정적으로 작동합니다.

 

Publisher Confirms의 한계

1. 컨슈머 처리 여부는 보장하지 않음
   브로커(Exchange 또는 Queue)에 메시지가 도달했는지는 확인할 수 있지만, 실제로 컨슈머가 메시지를 소비하고 비즈니스 로직을 수행했는지 여부는 알 수 없습니다.
2. 중복 전송 가능성 존재
   NACK 응답 시 동일 메시지를 재전송하는 과정에서 중복 메시지가 발생할 수 있으며, 이를 방지하기 위해 idempotent(멱등성) 을 갖춘 Consumer 설계가 필요합니다.
3. 지속성 설정에 따른 성능 영향
   메시지에 deliveryMode = 2 (PERSISTENT) 설정을 적용하면 디스크에 안전하게 저장되지만, 그에 따른 전송 성능 저하가 일부 발생할 수 있습니다.

---

Producer 트랜잭션 전략 코드

1. 트랜잭션 메시징 (Transaction Messaging)

1-1. 트랜잭션 메시징 RabbitMQConfig.Class

@Configuration
public class RabbitMQConfig {

    @Bean
    public Queue queue() {
        return new Queue("transactionMessageQueue", true);
    }

    @Bean
    public MessageConverter messageConverter() {
        return new Jackson2JsonMessageConverter();
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMessageConverter(messageConverter);

        // 트랜잭션 활성화
        rabbitTemplate.setChannelTransacted(true);  

        return rabbitTemplate;
    }

}

 

위 코드는 RabbitMQ 설정에 해당하는 코드 입니다. 다른 코드는 크게 볼 필요가 없으며, 가장 중요한 부분은 RabbitTemplate에서 트랜잭션을 활성화(rabbitTemplate.setChannelTransacted(true)) 해준 것이 가장 중요한 부분 입니다.

 

 

1-2. 트랜잭션 메시징 Producer.Class

@Component
public class MessageProducer {
    private final StockRepository stockRepository;
    private final RabbitTemplate rabbitTemplate;

    public MessageProducer(StockRepository stockRepository, RabbitTemplate rabbitTemplate) {
        this.stockRepository = stockRepository;
        this.rabbitTemplate = rabbitTemplate;
    }

    @Transactional
    public void sendMessage(StockEntity stock, String testCase) {
        rabbitTemplate.execute(channel -> {
            try {
            
                // 트랜잭션 시작
                channel.txSelect();
                
                // JPA 저장
                StockEntity stockSaved = stockRepository.save(stock);

                // 메시지 발행
                rabbitTemplate.convertAndSend("transactionMessageQueue", stockSaved);

                // 임의 오류 발생
                if ("fail".equalsIgnoreCase(testCase)) {
                    throw new RuntimeException("트랜잭션 작업중에 에러 발생");
                }

                // 트랜잭션 커밋
                channel.txCommit();
            } catch (Exception e) {
                // 롤백
                channel.txRollback();
                throw new RuntimeException("트랜잭션 롤백 완료 ", e);
            } finally {
                if (channel != null) {
                    try {
                        channel.close();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
            return null;
        });
    }
}

 

위 코드는 Producer, 메시지 발행에 해당하는 부분 입니다. 메시지 발행의 순서는 다음과 같습니다.

 

• 메시지 발행 성공
  1. 트랜잭션 시작
  2. DB 저장
  3. 메시지 발행
  4. 트랜잭션 커밋
• 메시지 발행 실패
  1. 트랜잭션 시작
  2. DB 저장
  3. 메시지 발행
  4. 임의 에러 발생
  5. 트랜잭션 롤백

---

2. Publisher Confirms