Springboot + RabbitMQ DLX 사용하기

DLX 란?

 

Dead Letter Exchange 의 줄인말임.

 

메세지큐는 큐안에 메세지가 들어있을때 무조건 소비를 해줘야한다.

 

어떠한 이유로 메세지가 소비가 안되고 계속 남아있을때, 무한히 요청을 한다.

 

무한히 요청이 되면, 과부하로 인해 메세지큐서버가 다운될 수 있다.

 

그걸 방지하기위해 메세지 소비가 안될 상황에 흔히 부르는 DLX 큐를 생성해 메세지를 옮겨주는 장소이다.

 

즉. 메세지소비 시 오류가 생겼을때, 임시로 대피시켜주는 장소라고 보면 된다.

 

 

바로 사용해보자

local에 rabbitmq를 띄어 놓기 (도커)

docker run -d --name rabbitmq -p 5672:5672 -p 15672:15672 --restart=unless-stopped -e RABBITMQ_DEFAULT_USER=root -e RABBITMQ_DEFAULT_PASS=root rabbitmq:management

 

의존성 주입

implementation ("org.springframework.boot:spring-boot-starter-amqp")

 

application.yml

spring:
  rabbitmq:
    host: localhost
    port: 5672
    username: root
    password: root

 

RedisConfig.java

 

"sendMailQueue" 이름을 가진 큐 생성 (dlx 설정부분이다.)

 

x-dead-letter-exchange를 mail.dlx.exchange 로 설정하였다.

메세지가 제대로 소비하지못하면, mail.dlx.exchange로 가라! 라는 설정이다.

또한 mail.dlx.exchange가 mail.dlx.queue 와 바인딩이 되어있기에,
mail.dlx.exchange에 들어온 메세지가  mail.dlx.queue로 전달해준다.

그리고 mail.dlx.queue에 쌓이게 된다.

@Bean
public Queue createSendMailQueue() {
    Queue queue = new Queue("sendMailQueue", true);
    queue.addArgument("x-dead-letter-exchange", "mail.dlx.exchange");
    queue.addArgument("x-dead-letter-routing-key", "mail.dlx.queue");

    return queue;
}

"sendMailQueue.exchange" 이름을 가진 exchange 생성

@Bean
public DirectExchange createSendMailDirectExchange() {
    return new DirectExchange("sendMailQueue.exchange", true, false, null);
}

 

위에 둘을 바인딩.

@Bean
public Binding createSendMailBinding() {
    return BindingBuilder.bind(createSendMailQueue()).to(createSendMailDirectExchange()).withQueueName();
}

 

----------------------------------------------------------------------------------------------------------------------

dlx 을 위해 dlx 큐를 생성

 

@Bean
public Binding createSendMailDlxBinding() {
    return BindingBuilder.bind(mailDlxQueue()).to(mailDlxExchange()).withQueueName();
}

 

exchange는 기존 만들어놓은 exchange 사용 / 바인딩 완료

@Bean
public Binding createSendSmsDlxBinding() {
    return BindingBuilder.bind(smsDlxQueue()).to(smsDlxExchange()).withQueueName();
}

 

이제 모든 세팅이 완료됐다.

 

 

테스트해보자

@RabbitListener(queues = "sendMailQueue")
    public void receiveData(MailReqDto mailReqDto) throws Exception {
        log.info(" ::: 메세지 큐로 부터 받았습니다. 받은정보 : {} :::", mailReqDto);
        try {
            webClient.webClientMail()
                    .post()
                    .uri("/mails")
                    .contentType(MediaType.APPLICATION_JSON)
                    .body(BodyInserters.fromValue(new NaverMailingReqDto(agreeInfoSidId, Collections.singletonList(new RecipientForRequest(mailReqDto, new Parameters(mailReqDto))))))
                    .retrieve()
                    .bodyToMono(NaverMailingResDto.class)
                    .timeout(Duration.ofMinutes(5))
                    .subscribe(
                            reseponse -> {
                                log.info("::: 메세지 전송 완료입니다. :::");
                                log.info("결과 ::: {}", reseponse);

                                amqpTemplate.convertAndSend("sendMailLogQueue", new MailResDto(reseponse,mailReqDto));
                                log.info(" ::: 메세지 큐에 전송하였습니다. 담은정보 : {} :::", new MailResDto(reseponse,mailReqDto));
                            },
                            error -> {
                                log.error(":::메세지 전송 실패::: : {}",error);
                                throw new MtnCommonException(MtnErrorCode.GENERAL_FAIL.getCode(), MtnErrorCode.GENERAL_FAIL.getMessage());
                            }
                    );
        } catch (Exception e) {
            throw new AmqpRejectAndDontRequeueException("failed to proccess message :" + mailReqDto, e);
        }
    }

 

위의 코드를 보자.

 

NCP mailing service 외부API  요청 코드이다.

 

try , catch 문으로 묶어 catch로 빠질 시 메세지큐에서 제공하는 AmqpRejectAndDontRequeueExcpetion() 을 사용하였다.

 

이 throw는 메세지가 소비하지 못할 시 reject 판단처리해주는 녀석이다.

 

reject으로 인식했기때문에, 메세지큐는 설정해준대로 dlx 큐로 가게된다.

 

 

 

처리되지못하고 reject된 메세지는 dlx로 이동하여 위에 그림과 같이 쌓이게 된다.

 

그 이후 후처리를 작성해주면 된다.