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>
</dependenciesapplication.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=manualpackage 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.BookHandlerLast updated