RabbitMQ 지연 큐(delayed queue)

delayed queue 큐는 말그대로 메시지 보낸 바로 즉시 받는게 아니라 나중에 받는다는 의미.

지연 큐 사용적합한 상황 아래와 같은 경우일것 같다.

• 주문기능 : 일정한 시간 지나도 최종 결제를 하지 않으면 주문 자동취소 되게 하려는 경우.

• 메시지 기능 : 주문완료후 60초 이내 고객에 메시지를 전송하도록 하고싶은 경우.

• 실패에 대한 재시도 : 업무시도 실패 후 일정한 시간후 재시도 하게끔 하는 경우.

이외 에도 많은 곳에서 지연 큐를 사용할수 있을것으로 추정된다.

RabbitMQ 자체는 지연 기능이 없지만 Time-To-Live Extensions 과 Dead Letter Exchange 의 특징을 이용하여 지연큐를 기능을 구현해 낼수 있다.

pom.xml 에 spring-boot-starter-amqp dependency 추가

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
    </dependency>
    <dependency>
        <groupId>com.alibaba</groupId>
        <artifactId>fastjson</artifactId>
        <version>1.2.46</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies

application.properties 에 rabbitmq 설정을 추가 한다.

spring.rabbitmq.username=demo
spring.rabbitmq.password=demo
#혹은 기타 ip
spring.rabbitmq.host=127.0.0.1  
spring.rabbitmq.port=5672
spring.rabbitmq.virtual-host=/
# 수동ACK , 자동 ACK모드 활성화 하지 않는다. 목적은 에러 발생후 message 유실되는걸 방지. 기본 값 : none
spring.rabbitmq.listener.simple.acknowledge-mode=manual

package com.demo.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMQ설정
 *
 * @author Blake
 * @since 2022/12/28 0011
 */
@Configuration
public class RabbitConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitConfig.class);

    @Bean
    public RabbitTemplate rabbitTemplate(CachingConnectionFactory connectionFactory) {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause));
        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message));
        return rabbitTemplate;
    }

    /**
     * 지연큐 TTL 이름
     */
    private static final String REGISTER_DELAY_QUEUE = "dev.book.register.delay.queue";
    /**
     * DLX，dead letter exchange 전송
     * TODO 여기 exchange 매우중요한 부분 , 메시지가 여기로 보내진다.
     */
    public static final String REGISTER_DELAY_EXCHANGE = "dev.book.register.delay.exchange";
    /**
     * routing key 이름
     * TODO 여기 routingKey 매우중요한 부분, 메시는  routingKey 보내진다.
     */
    public static final String DELAY_ROUTING_KEY = "";


    public static final String REGISTER_QUEUE_NAME = "dev.book.register.queue";
    public static final String REGISTER_EXCHANGE_NAME = "dev.book.register.exchange";
    public static final String ROUTING_KEY = "all";

    /**
     * 지연큐에 대한 설정
     * <p>
     * 1、params.put("x-message-ttl", 5 * 1000);
     * TODO 방식1: 직접 Queue  지연시간 설정, 큐에 expaired된 시간을 설정하면? 별로 유연한 설정 방법은 아닌듯.
     * 2、rabbitTemplate.convertAndSend(book, message -> {
     * message.getMessageProperties().setExpiration(2 * 1000 + "");
     * return message;
     * });
     * TODO 방식2: 매번 메시지 전송시 동적으로 지연시간을 설정한다. 조금더 유연해 보인다.
     **/
    @Bean
    public Queue delayProcessQueue() {
        Map<String, Object> params = new HashMap<>();
        // x-dead-letter-exchange : DLX 이름 정의
        params.put("x-dead-letter-exchange", REGISTER_EXCHANGE_NAME);
        // x-dead-letter-routing-key : routing-key 이름 정의
        params.put("x-dead-letter-routing-key", ROUTING_KEY);
        return new Queue(REGISTER_DELAY_QUEUE, true, false, false, params);
    }

    /**
     * 큐를 exchange 에 바인딩시켜야 한다. 해당 메시지와 라우터와 매칭되어야 한다.
     * TODO TopicExchange 처럼 여러개를 매칭할수 없다.
     *
     * @return DirectExchange
     */
    @Bean
    public DirectExchange delayExchange() {
        return new DirectExchange(REGISTER_DELAY_EXCHANGE);
    }

    @Bean
    public Binding dlxBinding() {
        return BindingBuilder.bind(delayProcessQueue()).to(delayExchange()).with(DELAY_ROUTING_KEY);
    }


    @Bean
    public Queue registerBookQueue() {
        return new Queue(REGISTER_QUEUE_NAME, true);
    }

  
    @Bean
    public TopicExchange registerBookTopicExchange() {
        return new TopicExchange(REGISTER_EXCHANGE_NAME);
    }

    @Bean
    public Binding registerBookBinding() {
       
        return BindingBuilder.bind(registerBookQueue()).to(registerBookTopicExchange()).with(ROUTING_KEY);
    }

}

Book (demo) 클래스를 만든다.

public class Book implements java.io.Serializable {

    private static final long serialVersionUID = -2164058270260403154L;

    private String id;
    private String name;
	// get set ...생략
}

controller 작성

package com.demo.controller;

import com.demo.config.RabbitConfig;
import com.demo.entity.Book;
import com.demo.handler.BookHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.AbstractJavaTypeMapper;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;

import java.time.LocalDateTime;

/**
 * @author Blake
 * @since 2022/12/28 0002
 */
@RestController
@RequestMapping(value = "/books")
public class BookController {

    private static final Logger log = LoggerFactory.getLogger(BookController.class);

    private final RabbitTemplate rabbitTemplate;

    @Autowired
    public BookController(RabbitTemplate rabbitTemplate) {
        this.rabbitTemplate = rabbitTemplate;
    }

    /**
     * this.rabbitTemplate.convertAndSend(RabbitConfig.REGISTER_DELAY_EXCHANGE, RabbitConfig.DELAY_ROUTING_KEY, book); 对应 {@link BookHandler#listenerDelayQueue}
     */
    @GetMapping
    public void defaultMessage() {
        Book book = new Book();
        book.setId("1");
        book.setName("하이!Spring Boot");
        // 큐추가
        this.rabbitTemplate.convertAndSend(RabbitConfig.REGISTER_DELAY_EXCHANGE, RabbitConfig.DELAY_ROUTING_KEY, book, message -> {
            // TODO 해도되고 안해도 되는 부분, 직접 처리 해야 됨
            message.getMessageProperties().setHeader(AbstractJavaTypeMapper.DEFAULT_CONTENT_CLASSID_FIELD_NAME, Book.class.getName());
            // TODO 만일  params.put("x-message-ttl", 5 * 1000); 설정하였다면 그러면 이 라인도 생략 가능.
            message.getMessageProperties().setExpiration(5 * 1000 + "");
            return message;
        });
        log.info("[전송시간] - [{}]", LocalDateTime.now());
    }
}

메시지 소비자(Consumer)

package com.demo.handler;

import com.demo.config.RabbitConfig;
import com.demo.entity.Book;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.time.LocalDateTime;

/**
 * BOOK_QUEUE 소비자
 *
 * @author Blake
 * @since 2022/12/28 0011
 */
@Component
public class BookHandler {

    private static final Logger log = LoggerFactory.getLogger(BookHandler.class);

    @RabbitListener(queues = {RabbitConfig.REGISTER_QUEUE_NAME})
    public void listenerDelayQueue(Book book, Message message, Channel channel) {
        log.info("[listenerManualAck 모니터링하고 있는 message] - [소비시간] - [{}] - [{}]", LocalDateTime.now(), book.toString());
        try {
            // TODO MQ에 메시지가 성공적으로 소비되었으니 ACK 해도 된다고 알려줌.
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
        } catch (IOException e) {
            // TODO 만일 실패했을때 다른 큐에 넣어서 처리하도록 한다.
        }
    }
}

메인 클래스

package com.demo;

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @author Blake
 */
@SpringBootApplication
public class RabbitMqDemoApplication {

    public static void main(String[] args) {
        SpringApplication.run(RabbitMqDemoApplication.class, args);
    }
}

RabbitMqDemoApplication 를 구동하면 아래와 같은 메시지가 출력 될것이다.

2022-12-28 18:06:36.248  INFO 29048 --- [nio-8080-exec-1] com.demo.controller.BookController     : [전송시간] - [2018-05-23T19:56:36.248]
2022-12-28 18:06:41.256  INFO 29048 --- [cTaskExecutor-1] com.demo.handler.BookHandler