SpringBoot基于RabbitMQ實現(xiàn)消息延時隊列的方案
知識小科普
在此之前，簡單說明下基于RabbitMQ實現(xiàn)延時隊列的相關(guān)知識及說明下延時隊列的使用場景。
延時隊列使用場景
在很多的業(yè)務(wù)場景中，延時隊列可以實現(xiàn)很多功能，此類業(yè)務(wù)中，一般上是非實時的，需要延遲處理的，需要進行重試補償?shù)摹?/p>
- 訂單超時關(guān)閉：在支付場景中，一般上訂單在創(chuàng)建后30分鐘或1小時內(nèi)未支付的，會自動取消訂單。
- 短信或者郵件通知：在一些注冊或者下單業(yè)務(wù)時，需要在1分鐘或者特定時間后進行短信或者郵件發(fā)送相關(guān)資料的。本身此類業(yè)務(wù)于主業(yè)務(wù)是無關(guān)聯(lián)性的，一般上的做法是進行異步發(fā)送。
- 重試場景：比如消息通知，在第一次通知出現(xiàn)異常時，會在隔幾分鐘之后進行再次重試發(fā)送。
RabbitMQ實現(xiàn)延時隊列
本身在RabbitMQ中是未直接提供延時隊列功能的，但可以使用 TTL(Time-To-Live，存活時間) 和 DLX(Dead-Letter-Exchange ，死信隊列交換機)的特性實現(xiàn)延時隊列的功能。
存活時間（Time-To-Live 簡稱 TTL）
RabbitMQ中可以對隊列和消息分別設(shè)置TTL，TTL表明了一條消息可在隊列中存活的最大時間。當某條消息被設(shè)置了TTL或者當某條消息進入了設(shè)置了TTL的隊列時，這條消息會在TTL時間后死亡成為Dead Letter。如果既配置了消息的TTL，又配置了隊列的TTL，那么較小的那個值會被取用。
死信交換（Dead Letter Exchanges 簡稱 DLX）
上個知識點也提到了，設(shè)置了 TTL 的消息或隊列最終會成為 Dead Letter ，當消息在一個隊列中變成死信之后，它能被重新發(fā)送到另一個交換機中，這個交換機就是DLX，綁定此DLX的隊列就是死信隊列。
一個消息變成死信一般上是由于以下幾種情況;
消息被拒絕
消息過期
隊列達到了最大長度。
所以，通過 TTL 和 DLX 的特性可以模擬實現(xiàn)延時隊列的功能。當隊列中的消息超時成為死信后，會把消息死信重新發(fā)送到配置好的交換機中，然后分發(fā)到真實的消費隊列。故簡單來說，我們可以創(chuàng)建2個隊列，一個隊列用于發(fā)送消息，一個隊列用于消息過期后的轉(zhuǎn)發(fā)的目標隊列。
SpringBoot集成RabbitMQ實現(xiàn)延時隊列實戰(zhàn)
以下使用 SpringBoot 集成 RabbitMQ 進行實戰(zhàn)說明，在進行 http 消息通知時，若通知失?。ǖ刂凡豢捎没蛘哌B接超時）時，將此消息轉(zhuǎn)入延時隊列中，待特定時間后進行重新發(fā)送。
0.引入pom依賴
<!-- rabbit -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
<!-- 簡化http操作 -->
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-http</artifactId>
<version>4.5.16</version>
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-json</artifactId>
<version>4.5.16</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
1.編寫rabbitmq配置文件（關(guān)鍵配置）RabbitConfig.java
/**
*
* @ClassName 類名：RabbitConfig
* @Description 功能說明：
* <p>
* TODO
*</p>
************************************************************************
* @date 創(chuàng)建日期：2019年7月17日
* @author 創(chuàng)建人：oKong
* @version 版本號：V1.0
*<p>
***************************修訂記錄*************************************
*
* 2019年7月17日 oKong 創(chuàng)建該類功能。
*
***********************************************************************
*</p>
*/
@Configuration
public class RabbitConfig {
@Autowired
ConnectionFactory connectionFactory;
/**
* 消費者線程數(shù) 設(shè)置大點 大概率是能通知到的
*/
@Value("${http.notify.concurrency:50}")
int concurrency;
/**
* 延遲隊列的消費者線程數(shù) 可設(shè)置小點
*/
@Value("${http.notify.delay.concurrency:20}")
int delayConcurrency;
@Bean
public RabbitAdmin rabbitAdmin() {
return new RabbitAdmin(connectionFactory);
}
@Bean
public DirectExchange httpMessageNotifyDirectExchange(RabbitAdmin rabbitAdmin) {
//durable 是否持久化
//autoDelete 是否自動刪除，即服務(wù)端或者客服端下線后 交換機自動刪除
DirectExchange directExchange = new DirectExchange(ApplicationConstant.HTTP_MESSAGE_EXCHANGE,true,false);
directExchange.setAdminsThatShouldDeclare(rabbitAdmin);
return directExchange;
}
//設(shè)置消息隊列
@Bean
public Queue httpMessageStartQueue(RabbitAdmin rabbitAdmin) {
/*
創(chuàng)建接收隊列，4個參數(shù)
name - 隊列名稱
durable - false，不進行持有化
exclusive - true，獨占性
autoDelete - true，自動刪除*/
Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME, true, false, false);
queue.setAdminsThatShouldDeclare(rabbitAdmin);
return queue;
}
//隊列綁定交換機
@Bean
public Binding bindingStartQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageStartQueue) {
Binding binding = BindingBuilder.bind(httpMessageStartQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_START_RK);
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}
@Bean
public Queue httpMessageOneQueue(RabbitAdmin rabbitAdmin) {
Queue queue = new Queue(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME, true, false, false);
queue.setAdminsThatShouldDeclare(rabbitAdmin);
return queue;
}
@Bean
public Binding bindingOneQuene(RabbitAdmin rabbitAdmin,DirectExchange httpMessageNotifyDirectExchange, Queue httpMessageOneQueue) {
Binding binding = BindingBuilder.bind(httpMessageOneQueue).to(httpMessageNotifyDirectExchange).with(ApplicationConstant.HTTP_MESSAGE_ONE_RK);
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}
//-------------設(shè)置延遲隊列--開始--------------------
@Bean
public Queue httpDelayOneQueue() {
//name - 隊列名稱
//durable - true
//exclusive - false
//autoDelete - false
return QueueBuilder.durable("http.message.dlx.one")
//以下是重點：當變成死信隊列時，會轉(zhuǎn)發(fā)至 路由為x-dead-letter-exchange及x-dead-letter-routing-key的隊列中
.withArgument("x-dead-letter-exchange", ApplicationConstant.HTTP_MESSAGE_EXCHANGE)
.withArgument("x-dead-letter-routing-key", ApplicationConstant.HTTP_MESSAGE_ONE_RK)
.withArgument("x-message-ttl", 1*60*1000)//1分鐘 過期時間（單位：毫秒），當過期后 會變成死信隊列，之后進行轉(zhuǎn)發(fā)
.build();
}
//綁定到交換機上
@Bean
public Binding bindingDelayOneQuene(RabbitAdmin rabbitAdmin, DirectExchange httpMessageNotifyDirectExchange, Queue httpDelayOneQueue) {
Binding binding = BindingBuilder.bind(httpDelayOneQueue).to(httpMessageNotifyDirectExchange).with("delay.one");
binding.setAdminsThatShouldDeclare(rabbitAdmin);
return binding;
}
//-------------設(shè)置延遲隊列--結(jié)束--------------------
//建議將正常的隊列和延遲處理的隊列分開
//設(shè)置監(jiān)聽容器
@Bean("notifyListenerContainer")
public SimpleRabbitListenerContainerFactory httpNotifyListenerContainer() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手動ack
factory.setConnectionFactory(connectionFactory);
factory.setPrefetchCount(1);
factory.setConcurrentConsumers(concurrency);
return factory;
}
// 設(shè)置監(jiān)聽容器
@Bean("delayNotifyListenerContainer")
public SimpleRabbitListenerContainerFactory httpDelayNotifyListenerContainer() {
SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
factory.setAcknowledgeMode(AcknowledgeMode.MANUAL);// 手動ack
factory.setConnectionFactory(connectionFactory);
factory.setPrefetchCount(1);
factory.setConcurrentConsumers(delayConcurrency);
return factory;
}
}
ApplicationConstant.java
public class ApplicationConstant {
/**
* 發(fā)送http通知的 exchange 隊列
*/
public static final String HTTP_MESSAGE_EXCHANGE = "http.message.exchange";
/**
* 配置消息隊列和路由key值
*/
public static final String HTTP_MESSAGE_START_QUEUE_NAME = "http.message.start";
public static final String HTTP_MESSAGE_START_RK = "rk.start";
public static final String HTTP_MESSAGE_ONE_QUEUE_NAME = "http.message.one";
public static final String HTTP_MESSAGE_ONE_RK = "rk.one";
/**
* 通知隊列對應(yīng)的延遲隊列關(guān)系，即過期隊列之后發(fā)送到下一個的隊列信息,可以根據(jù)實際情況添加，當然也可以根據(jù)一定規(guī)則自動生成
*/
public static final Map<String,String> delayRefMap = new HashMap<String, String>() {
/**
*
*/
private static final long serialVersionUID = -779823216035682493L;
{
put(HTTP_MESSAGE_START_QUEUE_NAME, "delay.one");
}
};
}
簡單來說，就是創(chuàng)建一個正常消息發(fā)送隊列，用于接收http消息請求的參數(shù)，同時進行http請求。同時，創(chuàng)建一個延時隊列，設(shè)置其 x-dead-letter-exchange 、x-dead-letter-routing-key 和
x-message-ttl 值，將其轉(zhuǎn)發(fā)到正常的隊列中。使用一個map對象維護一個關(guān)系，當正常消息異常時，需要發(fā)送的延時隊列的隊列名稱，當然時間場景匯總，根據(jù)需要可以進行動態(tài)配置或者根據(jù)一定規(guī)則進行動態(tài)映射。
2.創(chuàng)建監(jiān)聽類
用于消息的消費操作，此處使用@RabbitListener來消費消息（當然也可以使用SimpleMessageListenerContainer進行消息配置的），創(chuàng)建了一個正常消息監(jiān)聽和延時隊列監(jiān)聽，由于一般上異常通知是低概率事件，可根據(jù)不同的監(jiān)聽容器進行差異化配置。
/**
*
* @ClassName 類名：HttpMessagerLister
* @Description 功能說明：http通知消費監(jiān)聽接口
* <p>
* TODO
*</p>
************************************************************************
* @date 創(chuàng)建日期：2019年7月17日
* @author 創(chuàng)建人：oKong
* @version 版本號：V1.0
*<p>
***************************修訂記錄*************************************
*
* 2019年7月17日 oKong 創(chuàng)建該類功能。
*
***********************************************************************
*</p>
*/
@Component
@Slf4j
public class HttpMessagerLister {
@Autowired
HttpMessagerService messagerService;
@RabbitListener(id = "httpMessageNotifyConsumer", queues = {ApplicationConstant.HTTP_MESSAGE_START_QUEUE_NAME}, containerFactory = "notifyListenerContainer")
public void httpMessageNotifyConsumer(Message message, Channel channel) throws Exception {
doHandler(message, channel);
}
@RabbitListener(id= "httpDelayMessageNotifyConsumer", queues = {
ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME,}, containerFactory = "delayNotifyListenerContainer")
public void httpDelayMessageNotifyConsumer(Message message, Channel channel) throws Exception {
doHandler(message, channel);
}
private void doHandler(Message message, Channel channel) throws Exception {
String body = new String(message.getBody(),"utf-8");
String queue = message.getMessageProperties().getConsumerQueue();
log.info("接收到通知請求：{}，隊列名：{}",body, queue);
//消息對象轉(zhuǎn)換
try {
HttpEntity httpNotifyDto = JSONUtil.toBean(body, HttpEntity.class);
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
//發(fā)送通知
messagerService.notify(queue, httpNotifyDto);
} catch(Exception e) {
log.error(e.getMessage());
//ack
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}
}
}
HttpMessagerService.java ：消息真正處理的類，此類是關(guān)鍵，這里未進行日志記錄，真實場景中，強烈建議進行消息通知的日志存儲，防止日后信息的查看，同時也能通過發(fā)送狀態(tài)，在重試次數(shù)都失敗后，進行定時再次發(fā)送功能，同時也有據(jù)可查。
@Component
@Slf4j
public class HttpMessagerService {
@Autowired
AmqpTemplate mqTemplate;
public void notify(String queue,HttpEntity httpEntity) {
//發(fā)起請求
log.info("開始發(fā)起http請求:{}", httpEntity);
try {
switch(httpEntity.getMethod().toLowerCase()) {
case "POST":
HttpUtil.post(httpEntity.getUrl(), httpEntity.getParams());
break;
case "GET":
default:
HttpUtil.get(httpEntity.getUrl(), httpEntity.getParams());
}
} catch (Exception e) {
//發(fā)生異常，放入延遲隊列中
String nextRk = ApplicationConstant.delayRefMap.get(queue);
if(ApplicationConstant.HTTP_MESSAGE_ONE_QUEUE_NAME.equals(queue)) {
//若已經(jīng)是最后一個延遲隊列的消息隊列了，則后續(xù)可直接放入數(shù)據(jù)庫中 待后續(xù)定時策略進行再次發(fā)送
log.warn("http通知已經(jīng)通知N次失敗，進入定時進行發(fā)起通知,url={}", httpEntity.getUrl());
} else {
log.warn("http重新發(fā)送通知：{}, 通知隊列rk為：{}, 原隊列：{}", httpEntity.getUrl(), nextRk, queue);
mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, nextRk, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
}
}
}
}
3.創(chuàng)建控制層服務(wù)（真實場景中，如SpringCloud微服務(wù)中，一般上是創(chuàng)建個api接口，供其他服務(wù)進行調(diào)用）
@Slf4j
@RestController
@Api(tags = "http測試接口")
public class HttpDemoController {
@Autowired
AmqpTemplate mqTemplate;
@PostMapping("/send")
@ApiOperation(value="send",notes = "發(fā)送http測試")
public String sendHttp(@RequestBody HttpEntity httpEntity) {
//發(fā)送http請求
log.info("開始發(fā)起http請求，發(fā)布異步消息：{}", httpEntity);
mqTemplate.convertAndSend(ApplicationConstant.HTTP_MESSAGE_EXCHANGE, ApplicationConstant.HTTP_MESSAGE_START_RK, cn.hutool.json.JSONUtil.toJsonStr(httpEntity));
return "發(fā)送成功：url=" + httpEntity.getUrl();
}
}
4.配置文件添加RabbitMQ相關(guān)配置信息
spring.rabbitmq.host=127.0.0.1 spring.rabbitmq.port=5672 spring.rabbitmq.username=guest spring.rabbitmq.password=guest spring.rabbitmq.virtual-host=/ # 通知-消費者線程數(shù) 設(shè)置大點 大概率是能通知到的 http.notify.concurrency=150 # 延遲隊列的消費者線程數(shù) 可設(shè)置小點 http.notify.delay.concurrency=10
5.編寫啟動類。
@SpringBootApplication
@Slf4j
public class DelayQueueApplication {
public static void main(String[] args) throws Exception {
SpringApplication.run(DelayQueueApplication.class, args);
log.info("spring-boot-rabbitmq-delay-queue-chapter38服務(wù)啟動!");
}
}
6.啟動服務(wù)。使用swagger進行簡單調(diào)用測試。
正常通知：
2019-07-20 23:52:23.792 INFO 65216 --- [nio-8080-exec-1] c.l.l.s.c.controller.HttpDemoController : 開始發(fā)起http請求，發(fā)布異步消息：HttpEntity(url=www.baidu.com, params={a=1}, method=get)
2019-07-20 23:52:23.794 INFO 65216 --- [TaskExecutor-97] c.l.l.s.chapter38.mq.HttpMessagerLister : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com"}，隊列名：http.message.start
2019-07-20 23:52:23.794 INFO 65216 --- [TaskExecutor-97] c.l.l.s.c.service.HttpMessagerService : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com, params={a=1}, method=get)
異常通知：訪問一個不存在的地址
2019-07-20 23:53:14.699 INFO 65216 --- [nio-8080-exec-4] c.l.l.s.c.controller.HttpDemoController : 開始發(fā)起http請求，發(fā)布異步消息：HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.705 INFO 65216 --- [TaskExecutor-84] c.l.l.s.chapter38.mq.HttpMessagerLister : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，隊列名：http.message.start
2019-07-20 23:53:14.705 INFO 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:53:14.706 WARN 65216 --- [TaskExecutor-84] c.l.l.s.c.service.HttpMessagerService : http重新發(fā)送通知：www.baidu.com1, 通知隊列rk為：delay.one, 原隊列：http.message.start
在 RabbitMQ 后臺中，可以看見 http.message.dlx.one 隊列中存在這需要延時處理的消息，在一分鐘后會轉(zhuǎn)發(fā)至 http.message.one 隊列中。
在一分鐘后，可以看見消息本再次消費了。
2019-07-20 23:54:14.722 INFO 65216 --- [TaskExecutor-16] c.l.l.s.chapter38.mq.HttpMessagerLister : 接收到通知請求：{"method":"get","params":{"a":1},"url":"www.baidu.com1"}，隊列名：http.message.one
2019-07-20 23:54:14.723 INFO 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService : 開始發(fā)起http請求:HttpEntity(url=www.baidu.com1, params={a=1}, method=get)
2019-07-20 23:54:14.723 WARN 65216 --- [TaskExecutor-16] c.l.l.s.c.service.HttpMessagerService : http通知已經(jīng)通知N次失敗，進入定時進行發(fā)起通知,url=www.baidu.com1
一些最佳實踐
在正式場景中，一般上補償或者重試機制大概率是不會發(fā)送的，倘若發(fā)生時，一般上是第三方業(yè)務(wù)系統(tǒng)出現(xiàn)了問題，故一般上在進行補充時，應(yīng)該在非高峰期進行操作，故應(yīng)該對延時監(jiān)聽器，應(yīng)該在高峰期時停止消費，在非高峰期時進行消費。同時，還可以根據(jù)不同的通知類型，放入不一樣的延時隊列中，保障業(yè)務(wù)的正常。這里簡單說明下，動態(tài)停止或者啟動演示監(jiān)聽器的方式。一般上是使用RabbitListenerEndpointRegistry 對象獲取延時監(jiān)聽器，之后進行動態(tài)停止或者啟用?？稍O(shè)置 @RabbitListener 的id屬性，直接進行獲取，當然也可以直接獲取所有的監(jiān)聽器，進行自定義判斷了。
@Autowired
RabbitListenerEndpointRegistry registry;
@GetMapping("/set")
@ApiOperation(value = "set", notes = "設(shè)置消息監(jiān)聽器的狀態(tài)")
public String setSimpleMessageListenerContainer(String status) {
if("1".equals(status)) {
registry.getListenerContainer("httpDelayMessageNotifyConsumer").start();
} else {
registry.getListenerContainer("httpDelayMessageNotifyConsumer").stop();
}
return status;
}
這里，只是簡單進行演示說明，在真實場景下，可以使用定時器，判斷當前是否為高峰期，進而進行動態(tài)設(shè)置監(jiān)聽器的狀態(tài)。
參考資料
以上就是SpringBoot基于RabbitMQ實現(xiàn)消息延遲隊列的方案的詳細內(nèi)容，更多關(guān)于SpringBoot RabbitMQ消息延遲隊列的資料請關(guān)注腳本之家其它相關(guān)文章！
相關(guān)文章
springmvc配置線程池Executor做多線程并發(fā)操作的代碼實例
今天小編就為大家分享一篇關(guān)于springmvc配置線程池Executor做多線程并發(fā)操作的代碼實例，小編覺得內(nèi)容挺不錯的，現(xiàn)在分享給大家，具有很好的參考價值，需要的朋友一起跟隨小編來看看吧2019-03-03
這篇文章主要介紹了Java System.exit()退出程序方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2024-06-06
java中g(shù)et()方法和set()方法的作用淺析
這篇文章主要給大家介紹了關(guān)于java中g(shù)et()方法和set()方法的作用,set是是對數(shù)據(jù)進行設(shè)置,而get是對數(shù)據(jù)進行獲取,文中通過實例代碼介紹的非常詳細,需要的朋友可以參考下2023-07-07
調(diào)用Mybatis?plus中的saveBatch方法報找不到表的問題
在用Mybatis plus開發(fā)的項目中，用自帶的API批量保存的方法saveBatch操作時，發(fā)現(xiàn)報沒有找到表的錯誤，本文就來詳細的介紹一下解決方法，感興趣的可以了解一下2024-03-03
這篇文章主要介紹了springboot中的多個application文件，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-09-09
深入解析Java編程中的StringBuffer與StringBuider
這篇文章主要介紹了Java編程中的StringBuffer與StringBuider,是Java入門學習中的基礎(chǔ)知識,需要的朋友可以參考下2015-09-09
全角：指一個字符占用兩個標準字符位置，而半角：指一字符占用一個標準的字符位置，在日常開發(fā)中經(jīng)常會遇到全角半角轉(zhuǎn)換的要求，下面這篇文章主要給大家介紹了關(guān)于java中字符進行全角半角轉(zhuǎn)換的相關(guān)資料，文中通過示例代碼介紹的非常詳細，需要的朋友可以參考下。2017-08-08
spring一個項目多個模塊聚合打包問題解決方案(最新推薦)
最近遇到個需求,針對后端解耦模塊較多的項目,想在云端啟動時簡潔些只啟動一個jar文件的情景,本文重點給大家介紹spring一個項目多個模塊聚合打包問題解決方案,感興趣的朋友一起看看吧2023-09-09
詳談springboot過濾器和攔截器的實現(xiàn)及區(qū)別
今天小編就為大家分享一篇詳談springboot過濾器和攔截器的實現(xiàn)及區(qū)別，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2018-08-08
本文主要介紹了JAVA刪除字符串固定下標字串的實現(xiàn)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-04-04