[Spring] 분산 시스템에서 효과적인 데이터 전달 방법 (2) - RabbitMQ

[이전 포스팅] RDB를 사용하는 애플리케이션에서 효과적인 데이터 전달 방법

2023.11.04 - [BackEnd/Spring] - [Spring] 분산 시스템에서 효과적인 데이터 전달 방법 (1) - Transaction Outbox Pattern & Polling Publisher Pattern

 

2. RabbitMQ를 사용한 전달 방법

RabbitMQ는 기본적으로 AMQP(Advanced Mssage Queueing Protocol)를 구현한 브로커다.

 

예전에 공부했던 Publish/Subscribe 방식을 지원한다.

 

또한, RabbitMQ는 ACK(Acknowledgement) 메커니즘을 갖고있다.

 

RabbitMQ Publisher Confirms

Publisher가 메세지를  Queue에 정상적으로 발행했다면 Producer Confirm을,

 

Consumer가 메세지를 잘 받았다면 Consumer Ack을 보내게 된다.

 

1. Producer Confirm

 

Producer Confirm

 

Publisher가 메세지를 발행하면, Exchange가 메세지를 받아서 Queue에 저장한다.

 

이 과정이 완료되었다면 Exchange는 Publisher에게 ACK를 보내게 된다.

 

그렇다면 스프링에서 Producer Confirm을 확인할 수 있는 코드 예제를 보자.

 

@Service
public class MessagePublisher {

    public void sendMessage(CreateTaskEvent createTaskEvent) throws JsonProcessingException{
        String json = objectMapper.writeValueAsString(createTaskEvent);
        rabbitTemplate.send(EXCHANGE_NAME, 
                ROUTING_KEY, 
                new Message(json.getBytes(StandardCharsets.UTF_8)), 
                new CorrelationData(UUID.randomUUID().toString()));
    }
}

 

스프링에서 제공하는 CorrelationData 클래스는 Producer Confirm을 확인할 수 있는 기본 클래스이다.

 

위와 같이 RabbitTemplate을 통해 메세지를 보낼 때, CorrelationData객체를 UUID와같은 ID형태를 가지는 값을 생성하여 전달하면,

나중에 callback을 받을 수 있다.

 

@Configuration
@Slf4j
public class RabbitConfiguration {
	
    //...

    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory,
                                  MessageConverter messageConverter) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        
        //...
        
        
        //Producer Confirm
        rabbitTemplate.setConfirmCallback(((correlationData, ack, cause) -> {
            if (!ack) {
                Message message = correlationData.getReturned().getMessage();
                byte[] body = message.getBody();
                log.error("Fail to produce. ID: {}", correlationData.getId());
            }

        }));

        return rabbitTemplate;
    }

}

 

위와 같이 Configuration에서, rabbitTemplate에 setConfirmCallback메서드에 Producer Confirm 구현체를 넣어주면 된다.

 

여기서 전달되는 세 인자를 살펴보겠다.

 

1. correlationData

- 메세지를 발행할 때 생성한 객체

 

2. ack

- boolean이다. 성공/실패를 나타낸다.

 

3. cause

- 실패 원인을 담고있다.

 

위와 같이 콜백을 구현해두었다면 로그에서 correlationData 객체에 넣어두었던 ID를 확인할 수 있다.

 

위 Publisher Confirm을 사용하려면 설정을 해야하는데,

 

 

 

Spring Boot를 사용 중이라면,

 

● applicaiton.yml

spring:
  rabbitmq:
    publisher-confirm-type: correlated

 

 

Spring을 사용 중이라면, 아래와 같이 스프링 빈을 직접 정의해도 된다.

 

● RabbitConfiguration

@Bean
    public ConnectionFactory rabbitConnectionFactory() {
        CachingConnectionFactory cf = new CachingConnectionFactory();
        //...
        cf.setPublisherConfirmType(CachingConnectionFactory.ConfirmType.CORRELATED);
        return cf;
    }

 

 

 

2. Consumer Acknowledge

Consumer Acknowledge

 

Consumer가 Queue에 있는 Message를 정상적으로 가져왔다면 ACK를 보내는데, 이를 Consumer Acknowledge라고 한다.

 

Consumer Acknowledge를 구현하려면 Channel객체를 사용해야한다.

 

@Component
public class MessageListner {

    @RabbitListener(queues = "test.task")
    public void receiveMessage(Message message, Channel channel) throws IOException {

        //수동 ACK 전송
        channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);

        //수동 NACK 전송
        channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
    }
}

 

이 Channel객체는 메세지 큐와 애플리케이션의 연결을 추상화한 클래스이다.

이 클래스를 이용하면 Publish, Listen을 모두 사용할 수 있고, 직접 ACK와 NACK을 전송할 수 있는 메서드를 사용할 수 있다.

 

위 코드에서는 RabbitListner를 구현할 때, message외에 Channel객체를 같이 받으면 스프링 프레임워크에서 알맞은 객체를 전달해준다.

 

사실은 Channel 객체를 직접 받지 않더라도, 프레임워크가 Exception이 발생하면 자동으로 NACK를 보내주기 때문에 직접 구현할 필요는 없다.

 

3. Dead Letter Queue

그런데 만약, Consumer에서 계속 NACK을 보내는 상황이고, Queue에는 메세지가 계속 쌓이는 중이라면 어떻게 해야될까?

 

이 경우에, Dead Letter환경을 구축한다.

 

Dead Letter  Queue

 Queue에서 정상적으로 처리되지 못한 메세지들을 Dead Letter Queue에 넘긴다.

 

조건은 크게 아래 세 가지가 있다.

 

1. requeue = false이고, basic reject, basic NACK로 처리하는 경우

    requeue = true라면 메세지가 dead letter로 가지 않고 바로 원래 있던 queue로 되돌아간다.

 

2. queue의 TTL(Time To Live)이 넘어간 경우 (오래된 경우)

 

3. queue가 가득 차서 처리할 수 없는 상태가 되는 경우 

 

 

아래는 Dead Letter를 사용하기 위한 설정이다.

@Bean
public SimpleRabbitListenerContainerFactory retryContainerFactory(ConnectionFactory connectionFactory) {
    //...

    var containerFactory = new SimpleRabbitListenerContainerFactory();
    containerFactory.setConnectionFactory(connectionFactory);
    containerFactory.setAdviceChain(
            RetryInterceptorBuilder.stateless()
                    .maxAttempts(3)
                    .backOffOptions(1000, 2, 2000)
                    .recoverer(new RejectAndDontRequeueRecoverer())
                    .build()
    );
    return containerFactory;
}

 

위와 같이 설정해두면, 최대 3번 실행하고, retry 간격은 기본으로 1000(1초0), 간격의 배수는 2, 최대 간격은 2000(2초)가 된다.

 

아래 recoeverer에는 RejectAndDontRequeueRecoverer 클래스를 설정해두어야 Requeue를 하지 않는다.

 

 

아래는 Dead Letter Queue로 온 메세지들을 처리하는 예제이다.

@RabbitListener(
            queues = "test.deadletter",
            containerFactory = "deadLetterContainerFactory"
    )
    public void onDeadLetterMessage(Map<String, Object> rawEvent) {
        log.info("Dead Letter Message");
        // Alert failure or fallback
    }

 

RabbitListener로 메세지를 받으면 되고, 추가로 로그를 저장하거나 fallback을 시도하면 된다.

 

 

 

 

이로써 RabbitMQ를 사용하는 환경에서 Producer Confim을 확인하는 방법, Consumer ACK을 보내는 과정, Dead Letter Queue 환경 구축을 알아보았다.

다음 포스팅에서는 Kafka를 사용하는 환경에서 데이터 전달 환경을 구축해 보겠다.

 

 

 

 

 

 

참고
https://blog.rabbitmq.com/posts/2011/02/introducing-publisher-confirms/
https://spring.io/guides/gs/messaging-rabbitmq/
https://www.youtube.com/watch?v=uk5fRLUsBfk&t=1797s&ab_channel=NHNCloud