關(guān)于SpringBoot整合RabbitMQ實(shí)現(xiàn)死信隊(duì)列
更新時(shí)間：2023年05月20日 10:54:31 作者：叫我二蛋
這篇文章主要介紹了關(guān)于SpringBoot整合RabbitMQ實(shí)現(xiàn)死信隊(duì)列,死信隊(duì)列實(shí)際上就是一個(gè)普通的隊(duì)列，只是這個(gè)隊(duì)列跟死信交換機(jī)進(jìn)行了綁定，用來(lái)存放死信而已,需要的朋友可以參考下
概念介紹
什么是死信
死信可以理解成沒(méi)有被正常消費(fèi)的消息，在RabbitMQ中以下幾種情況會(huì)被認(rèn)定為死信：
- 消費(fèi)者使用basic.reject或basic.nack（重新排隊(duì)參數(shù)設(shè)置為false）對(duì)消息進(jìn)行否定確認(rèn)。
- 消息到達(dá)生存時(shí)間還未被消費(fèi)。
- 隊(duì)列超過(guò)長(zhǎng)度限制，消息被丟棄。
這些消息會(huì)被發(fā)送到死信交換機(jī)并路由到死信隊(duì)列中（在RabbitMQ中死信交換機(jī)和死信隊(duì)列就是普通的交換機(jī)和隊(duì)列）。其流轉(zhuǎn)過(guò)程如下圖
死信隊(duì)列應(yīng)用
- 作為消息可靠性的一個(gè)擴(kuò)展。比如，在隊(duì)列已滿的情況下也不會(huì)丟失消息。
- 可以實(shí)現(xiàn)延遲消費(fèi)功能。比如，訂單15分鐘內(nèi)未支付。
注意事項(xiàng)：基于死信隊(duì)列實(shí)現(xiàn)的延遲消費(fèi)不適合時(shí)間過(guò)于復(fù)雜的場(chǎng)景。比如，一個(gè)隊(duì)列中第一條消息TTL為10s，第二條消息TTL為5s，由于RabbitMQ只會(huì)監(jiān)聽(tīng)第一條消息，所以本應(yīng)第二條消息先達(dá)到TTL會(huì)在第一條消息的TTL之后。對(duì)于該現(xiàn)象有兩種解決方案：
- 維護(hù)多個(gè)隊(duì)列，每個(gè)隊(duì)列維護(hù)一個(gè)TTL時(shí)間。
- 使用延遲交換機(jī)。這種方式需要下載插件支持
工程搭建
環(huán)境說(shuō)明
- RabbitMQ環(huán)境
- Java版本：JDK1.8
- Maven版本：apache-maven-3.6.3
- 開(kāi)發(fā)工具：IntelliJ IDEA
搭建步驟
1.創(chuàng)建SpringBoot項(xiàng)目。
2.pom.xml文件導(dǎo)入RabbitMQ依賴。
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>3.application.yml文件添加RabbitMQ配置。
spring:
# rabbitmq配置信息 RabbitProperties類
rabbitmq:
host: 127.0.0.1
port: 5672
username: guest
password: guest
virtual-host: /
# 開(kāi)啟confirm機(jī)制
publisher-confirm-type: correlated
# 開(kāi)啟return機(jī)制
publisher-returns: true
#全局配置，局部配置存在就以局部為準(zhǔn)
listener:
simple:
acknowledge-mode: manual # 手動(dòng)ACK實(shí)現(xiàn)死信
準(zhǔn)備Exchange&Queue
@Configuration
public class RabbitMQConfig {
/**
* 正常隊(duì)列
*/
public static final String EXCHANGE = "boot-exchange";
public static final String QUEUE = "boot-queue";
public static final String ROUTING_KEY = "boot-rout";
/**
* 死信隊(duì)列
*/
public static final String DEAD_EXCHANGE = "dead-exchange";
public static final String DEAD_QUEUE = "dead-queue";
public static final String DEAD_ROUTING_KEY = "dead-rout";
/**
* 聲明死信交換機(jī)
*
* @return
*/
@Bean
public Exchange deadExchange() {
return ExchangeBuilder.directExchange(DEAD_EXCHANGE).build();
}
/**
* 聲明死信隊(duì)列
*
* @return
*/
@Bean
public Queue deadQueue() {
return QueueBuilder.durable(DEAD_QUEUE).build();
}
/**
* 綁定死信的隊(duì)列和交換機(jī)
*
* @param deadExchange
* @param deadQueue
* @return
*/
@Bean
public Binding deadBind(Exchange deadExchange, Queue deadQueue) {
return BindingBuilder.bind(deadQueue).to(deadExchange).with(DEAD_ROUTING_KEY).noargs();
}
/**
* 聲明交換機(jī)，同channel.exchangeDeclare(EXCHANGE, BuiltinExchangeType.DIRECT);
*
* @return
*/
@Bean
public Exchange bootExchange() {
return ExchangeBuilder.directExchange(EXCHANGE).build();
}
/**
* 聲明隊(duì)列，同channel.queueDeclare(QUEUE, true, false, false, null);
* 綁定死信交換機(jī)及路由key
*
* @return
*/
@Bean
public Queue bootQueue() {
return QueueBuilder.durable(QUEUE)
.deadLetterExchange(DEAD_EXCHANGE)
.deadLetterRoutingKey(DEAD_ROUTING_KEY)
//聲明隊(duì)列屬性有更改時(shí)需要?jiǎng)h除隊(duì)列
//給隊(duì)列設(shè)置消息時(shí)長(zhǎng)
//.ttl(10000)
//隊(duì)列最大長(zhǎng)度
.maxLength(1)
.build();
}
/**
* 綁定隊(duì)列和交換機(jī),同 channel.queueBind(QUEUE, EXCHANGE, ROUTING_KEY);
*
* @param bootExchange
* @param bootQueue
* @return
*/
@Bean
public Binding bootBind(Exchange bootExchange, Queue bootQueue) {
return BindingBuilder.bind(bootQueue).to(bootExchange).with(ROUTING_KEY).noargs();
}
}監(jiān)聽(tīng)死信隊(duì)列
@RabbitListener(queues = RabbitMQConfig.DEAD_QUEUE)
public void listener_dead(String msg, Channel channel, Message message) throws IOException {
System.out.println("死信接收到消息" + msg);
System.out.println("唯一標(biāo)識(shí):" + message.getMessageProperties().getCorrelationId());
System.out.println("messageID:" + message.getMessageProperties().getMessageId());
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
}方式一、消費(fèi)者拒絕&否認(rèn)
- 拒絕消息
@RabbitListener(queues = RabbitMQConfig.QUEUE)
public void listener(String msg, Channel channel, Message message) throws IOException {
System.out.println("接收到消息" + msg);
channel.basicReject(message.getMessageProperties().getDeliveryTag(), false)
}- 否認(rèn)消息
@RabbitListener(queues = RabbitMQConfig.QUEUE)
public void listener(String msg, Channel channel, Message message) throws IOException {
System.out.println("接收到消息" + msg);
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, false);
}方式二、超過(guò)消息TTL 發(fā)送消息時(shí)設(shè)置TTL
@SpringBootTest
public class Publisher {
@Autowired
private RabbitTemplate template;
/**
* 5秒未被消費(fèi)會(huì)路由到死信隊(duì)列
*/
@Test
public void publish_expir() {
template.convertAndSend(RabbitMQConfig.EXCHANGE, RabbitMQConfig.ROUTING_KEY, "hello expir dead", message -> {
message.getMessageProperties().setExpiration("5000");
return message;
});
}
}- 設(shè)置隊(duì)列所有消息的TTL
更新RabbitMQConfig類中bootQueue() ，更新后需要?jiǎng)h除隊(duì)列，因?yàn)殛?duì)列屬性有更改。
@Bean
public Queue bootQueue() {
return QueueBuilder.durable(QUEUE)
.deadLetterExchange(DEAD_EXCHANGE)
.deadLetterRoutingKey(DEAD_ROUTING_KEY)
//聲明隊(duì)列屬性有更改時(shí)需要?jiǎng)h除隊(duì)列
//給隊(duì)列設(shè)置消息時(shí)長(zhǎng)
.ttl(10000)
.build();
}方式三、超過(guò)隊(duì)列長(zhǎng)度限制
設(shè)置隊(duì)列長(zhǎng)度限制，當(dāng)隊(duì)列長(zhǎng)度超過(guò)設(shè)置的閾值，消息便會(huì)路由到死信隊(duì)列。
@Bean
public Queue bootQueue() {
return QueueBuilder.durable(QUEUE)
.deadLetterExchange(DEAD_EXCHANGE)
.deadLetterRoutingKey(DEAD_ROUTING_KEY)
//聲明隊(duì)列屬性有更改時(shí)需要?jiǎng)h除隊(duì)列
.maxLength(1)
.build();
}代碼倉(cāng)庫(kù) 點(diǎn)我
到此這篇關(guān)于關(guān)于SpringBoot整合RabbitMQ實(shí)現(xiàn)死信隊(duì)列的文章就介紹到這了,更多相關(guān)RabbitMQ實(shí)現(xiàn)死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Springboot使用RabbitMq延遲隊(duì)列和死信隊(duì)列詳解
- Springboot使用Rabbitmq的延時(shí)隊(duì)列+死信隊(duì)列實(shí)現(xiàn)消息延期消費(fèi)
- springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)
- SpringBoot整合RabbitMQ實(shí)現(xiàn)延遲隊(duì)列和死信隊(duì)列
- springboot中RabbitMQ死信隊(duì)列的實(shí)現(xiàn)示例
- Springboot結(jié)合rabbitmq實(shí)現(xiàn)的死信隊(duì)列
- SpringBoot+RabbitMQ?實(shí)現(xiàn)死信隊(duì)列的示例
- SpringBoot整合RabbitMQ處理死信隊(duì)列和延遲隊(duì)列
- Springboot集成RabbitMQ死信隊(duì)列的實(shí)現(xiàn)
- SpringBoot集成RabbitMQ的方法(死信隊(duì)列)
- SpringBoot4.0整合RabbitMQ死信隊(duì)列詳解
相關(guān)文章
Spring線程池ThreadPoolExecutor配置并且得到任務(wù)執(zhí)行的結(jié)果
今天小編就為大家分享一篇關(guān)于Spring線程池ThreadPoolExecutor配置并且得到任務(wù)執(zhí)行的結(jié)果，小編覺(jué)得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來(lái)看看吧2019-03-03
ArrayList和LinkedList區(qū)別及使用場(chǎng)景代碼解析
這篇文章主要介紹了ArrayList和LinkedList區(qū)別及使用場(chǎng)景代碼解析，小編覺(jué)得還是挺不錯(cuò)的，具有一定借鑒價(jià)值，需要的朋友可以參考下2018-01-01
這篇文章主要介紹了Java抽象類的構(gòu)造模板模式用法,結(jié)合實(shí)例形式分析了java使用抽象類構(gòu)造模板模式相關(guān)操作技巧,需要的朋友可以參考下2021-09-09
class 是java的關(guān)鍵字,在聲明java類時(shí)使用,Class是java JDK提供的一個(gè)類,完整路徑為java.lang.Class,下面這篇文章主要給大家介紹了關(guān)于Java中class和Class區(qū)別的相關(guān)資料,需要的朋友可以參考下2022-04-04
之前工作需要,查了挺多種Excel轉(zhuǎn)PDF的方法,下面這篇文章主要給大家介紹了關(guān)于java將excel轉(zhuǎn)為pdf的相關(guān)資料,文中通過(guò)代碼介紹的非常詳細(xì),需要的朋友可以參考下2024-01-01
IntelliJ IDEA 2020.3.3現(xiàn)已發(fā)布!新增“受信任項(xiàng)目”功能
這篇文章主要介紹了IntelliJ IDEA 2020.3.3現(xiàn)已發(fā)布!新增“受信任項(xiàng)目”功能,本文給大家分享了idea2020.3.3激活碼的詳細(xì)破解教程，每種方法都很好用，使用idea2020.3以下所有版本，需要的朋友可以參考下2021-03-03
idea2020安裝MybatisCodeHelper插件的圖文教程
這篇文章主要介紹了idea2020安裝MybatisCodeHelper插件的方法，本文通過(guò)圖文并茂的形式給大家介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-09-09
java實(shí)現(xiàn)發(fā)送短信驗(yàn)證碼
這篇文章主要為大家詳細(xì)介紹了java實(shí)現(xiàn)發(fā)送短信驗(yàn)證碼，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2019-07-07
java使用google身份驗(yàn)證器實(shí)現(xiàn)動(dòng)態(tài)口令驗(yàn)證的示例
本篇文章主要介紹了java使用google身份驗(yàn)證器實(shí)現(xiàn)動(dòng)態(tài)口令驗(yàn)證的示例，具有一定的參考價(jià)值，有興趣的可以了解一下2017-08-08
Feign 是一個(gè)聲明式的偽RPC的REST客戶端，它用了基于接口的注解方式，很方便的客戶端配置，Spring Cloud 給 Feign 添加了支持Spring MVC注解，并整合Ribbon及Eureka進(jìn)行支持負(fù)載均衡，這篇文章主要介紹了微服務(wù)之Feign的介紹與使用,需要的朋友可以參考下2022-07-07