springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)
更新時(shí)間：2024年10月23日 08:36:45 作者：饅頭警告
死信是無(wú)法被消費(fèi)的消息,產(chǎn)生原因包括消息TTL過(guò)期、隊(duì)列最大長(zhǎng)度達(dá)到以及消息被拒絕且不重新排隊(duì),RabbitMQ的死信隊(duì)列機(jī)制能夠有效防止消息數(shù)據(jù)丟失,適用于訂單業(yè)務(wù)等場(chǎng)景,本文就來(lái)介紹一下
一、死信的概念
先從概念解釋上搞清楚這個(gè)定義，死信，顧名思義就是無(wú)法被消費(fèi)的消息，字面意思可以這樣理
解，一般來(lái)說(shuō)，producer 將消息投遞到 broker 或者直接到 queue 里了，consumer 從 queue 取出消息進(jìn)行消費(fèi)，但某些時(shí)候由于特定的原因?qū)е?queue 中的某些消息無(wú)法被消費(fèi)，這樣的消息如果沒(méi)有后續(xù)的處理，就變成了死信，有死信自然就有了死信隊(duì)列。
應(yīng)用場(chǎng)景:為了保證訂單業(yè)務(wù)的消息數(shù)據(jù)不丟失，需要使用到 RabbitMQ 的死信隊(duì)列機(jī)制，當(dāng)消息
消費(fèi)發(fā)生異常時(shí)，將消息投入死信隊(duì)列中.還有比如說(shuō): 用戶在商城下單成功并點(diǎn)擊去支付后在指定時(shí)間未支付時(shí)自動(dòng)失效
二、死信的來(lái)源
1、消息 TTL 過(guò)期
2、隊(duì)列達(dá)到最大長(zhǎng)度(隊(duì)列滿了，無(wú)法再添加數(shù)據(jù)到 mq 中)
3、消息被拒絕(basic.reject 或 basic.nack)并且 requeue=false.
三、死信實(shí)戰(zhàn)
1、代碼架構(gòu)圖
生產(chǎn)者正常情況下走的是普通的交換機(jī)，這個(gè)交換機(jī)的類型是 direct ，它和普通隊(duì)列之間的關(guān)系是一個(gè)叫 "zhangsan" 的路由 key， 正常情況下會(huì)被 C1 消費(fèi)。
但是發(fā)生了上面所說(shuō)的三種情況中的一種，成為了死信，然后被轉(zhuǎn)換到死信交換機(jī)中，這個(gè)死信交換機(jī)也是 direct 類型，它們之間的 routingKey 是 "lisi",然后就進(jìn)入了死信隊(duì)列，死信隊(duì)列由 C2 消費(fèi)。
2、消息 TTL 過(guò)期
（1）消費(fèi)者
// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機(jī)名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊(duì)列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機(jī)的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊(duì)列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過(guò)期時(shí)間
//arguments.put("x-message-ttl",100000);
// 正常隊(duì)列設(shè)置死信交換機(jī)
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設(shè)置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊(duì)列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機(jī)與普通隊(duì)列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機(jī)與死信的隊(duì)列
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));
};
CancelCallback cancelCallback = consumerTag->{};
channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
}
}// 消費(fèi)者 2
public class Comsumer02 {
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
};
CancelCallback cancelCallback = consumerTag->{};
channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
}
}
（2）生產(chǎn)者
// 死信隊(duì)列 生產(chǎn)者代碼
public class Producer {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設(shè)置 TTL 的時(shí)間
AMQP.BasicProperties properties = new AMQP.BasicProperties().
builder().expiration("10000").build();
for (int i = 1; i < 11; i++) {
String message = "info" + i;
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",properties,message.getBytes());
}
}
}
（3）結(jié)果展示
3、隊(duì)列達(dá)到最大長(zhǎng)度
（1）消費(fèi)者
// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機(jī)名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊(duì)列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機(jī)的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊(duì)列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過(guò)期時(shí)間
//arguments.put("x-message-ttl",100000);
// 正常隊(duì)列設(shè)置死信交換機(jī)
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設(shè)置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
// 設(shè)置正常隊(duì)列的長(zhǎng)度的限制
arguments.put("x-max-length",6);
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊(duì)列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機(jī)與普通隊(duì)列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機(jī)與死信的隊(duì)列
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));
};
CancelCallback cancelCallback = consumerTag->{};
channel.basicConsume(NORMAL_QUEUE,true,deliverCallback,cancelCallback);
}
}// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 2
public class Comsumer02 {
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
};
CancelCallback cancelCallback = consumerTag->{};
channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
}
}（2）生產(chǎn)者
// 死信隊(duì)列 生產(chǎn)者代碼
public class Producer {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設(shè)置 TTL 的時(shí)間
/*AMQP.BasicProperties properties = new AMQP.BasicProperties().
builder().expiration("10000").build();*/
for (int i = 1; i < 11; i++) {
String message = "info" + i;
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
}
}
}（3）結(jié)果展示
4、消息被拒
（1）消費(fèi)者
// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 1
public class Comsumer01 {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機(jī)名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊(duì)列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機(jī)的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊(duì)列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過(guò)期時(shí)間
//arguments.put("x-message-ttl",100000);
// 正常隊(duì)列設(shè)置死信交換機(jī)
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設(shè)置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
// 設(shè)置正常隊(duì)列的長(zhǎng)度的限制
// arguments.put("x-max-length",6);
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊(duì)列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機(jī)與普通隊(duì)列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機(jī)與死信的隊(duì)列
channel.queueBind(DEAD_QUEUE,DEAD_EXCHANGE,"lisi");
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
String msg = new String(message.getBody(),"UTF-8");
if (msg.equals("info5")){
System.out.println("Consumer01 接收的消息是： " + msg + ": 此消息是被 C1 拒絕的");
// 拒絕，且不放囧普通隊(duì)列
channel.basicReject(message.getEnvelope().getDeliveryTag(),false);
}else {
System.out.println("Consumer01 接收的消息是： " + new String(message.getBody(),"UTF-8"));
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
}
};
CancelCallback cancelCallback = consumerTag->{};
// 開啟手動(dòng)應(yīng)答
channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);
}
}// 死信隊(duì)列 實(shí)戰(zhàn)
// 消費(fèi)者 2
public class Comsumer02 {
// 死信隊(duì)列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
System.out.println("等待接收消息.....");
DeliverCallback deliverCallback = ( consumerTag, message) ->{
System.out.println("Consumer02 接收的消息是： " + new String(message.getBody(),"UTF-8"));
};
CancelCallback cancelCallback = consumerTag->{};
channel.basicConsume(DEAD_QUEUE,true,deliverCallback,cancelCallback);
}
}（2）生產(chǎn)者
// 死信隊(duì)列 生產(chǎn)者代碼
public class Producer {
// 普通交換機(jī)名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設(shè)置 TTL 的時(shí)間
/*AMQP.BasicProperties properties = new AMQP.BasicProperties().
builder().expiration("10000").build();*/
for (int i = 1; i < 11; i++) {
String message = "info" + i;
channel.basicPublish(NORMAL_EXCHANGE,"zhangsan",null,message.getBytes());
}
}
}（3）結(jié)果展示
到此這篇關(guān)于springboot整合RabbitMQ中死信隊(duì)列的實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)springboot RabbitMQ死信隊(duì)列內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！
您可能感興趣的文章:
- Springboot使用RabbitMq延遲隊(duì)列和死信隊(duì)列詳解
- Springboot使用Rabbitmq的延時(shí)隊(duì)列+死信隊(duì)列實(shí)現(xiàn)消息延期消費(fèi)
- SpringBoot整合RabbitMQ實(shí)現(xiàn)延遲隊(duì)列和死信隊(duì)列
- springboot中RabbitMQ死信隊(duì)列的實(shí)現(xiàn)示例
- Springboot結(jié)合rabbitmq實(shí)現(xiàn)的死信隊(duì)列
- 關(guān)于SpringBoot整合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)文章
詳解使用Jenkins部署Spring Boot項(xiàng)目
這篇文章主要介紹了詳解使用Jenkins部署Spring Boot，小編覺(jué)得挺不錯(cuò)的，現(xiàn)在分享給大家，也給大家做個(gè)參考。一起跟隨小編過(guò)來(lái)看看吧2017-11-11
Springboot微服務(wù)項(xiàng)目整合Kafka實(shí)現(xiàn)文章上下架功能
這篇文章主要介紹了Springboot微服務(wù)項(xiàng)目整合Kafka實(shí)現(xiàn)文章上下架功能,包括Kafka消息發(fā)送快速入門及相關(guān)功能引入，本文通過(guò)示例代碼給大家介紹的非常詳細(xì)，需要的朋友可以參考下2022-07-07
基于Jackson實(shí)現(xiàn)API接口數(shù)據(jù)脫敏的示例詳解
用戶的一些敏感數(shù)據(jù),例如手機(jī)號(hào)、郵箱、身份證等信息,在數(shù)據(jù)庫(kù)以明文存儲(chǔ),但在接口返回?cái)?shù)據(jù)給瀏覽器（或三方客戶端）時(shí),希望對(duì)這些敏感數(shù)據(jù)進(jìn)行脫敏,所以本文就給大家介紹以惡如何利用Jackson實(shí)現(xiàn)API接口數(shù)據(jù)脫敏,需要的朋友可以參考下2023-08-08
SpringBoot開發(fā)案例 分布式集群共享Session詳解
這篇文章主要介紹了SpringBoot開發(fā)案例 分布式集群共享Session詳解，在分布式系統(tǒng)中，為了提升系統(tǒng)性能，通常會(huì)對(duì)單體項(xiàng)目進(jìn)行拆分，分解成多個(gè)基于功能的微服務(wù)，可能還會(huì)對(duì)單個(gè)微服務(wù)進(jìn)行水平擴(kuò)展，保證服務(wù)高可用,需要的朋友可以參考下2019-07-07
Java使用Spire.XLS?for?Java實(shí)現(xiàn)Excel與CSV的相互轉(zhuǎn)換
在工作中,Excel?和?CSV?幾乎是繞不開的兩種數(shù)據(jù)格式,本篇指南將從工程角度分析?Excel?和?CSV?的不同點(diǎn)及各自適合的使用環(huán)境,并通過(guò)實(shí)際代碼示例展示它們之間的相互轉(zhuǎn)換,幫助你在真實(shí)項(xiàng)目中做出更合理、更高效的數(shù)據(jù)格式選擇2026-01-01
SpringBoot?+?Vue?+?ElementUI?實(shí)現(xiàn)?el-table?分頁(yè)功能(詳細(xì)步驟)
本文詳細(xì)介紹了使用SpringBoot和Vue.js結(jié)合ElementUI實(shí)現(xiàn)分頁(yè)功能的數(shù)據(jù)表格,從后端分頁(yè)邏輯到前端展示和狀態(tài)管理,全面解析如何高效處理大量數(shù)據(jù),提升用戶體驗(yàn)與系統(tǒng)性能,感興趣的朋友跟隨小編一起看看吧2024-09-09
SpringBoot整合screw實(shí)現(xiàn)自動(dòng)生成數(shù)據(jù)庫(kù)設(shè)計(jì)文檔
使用navicat工作的話，導(dǎo)出的格式是excel不符合格式，還得自己整理。所以本文將用screw工具包，整合到springboot的項(xiàng)目中便可以自動(dòng)生成數(shù)據(jù)庫(kù)設(shè)計(jì)文檔，非常方便，下面就分享一下教程2022-11-11
本文主要介紹了Spring Boot集成Druid查看配置是否生效的方法，文中通過(guò)示例代碼介紹的非常詳細(xì)，對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值，需要的朋友們下面隨著小編來(lái)一起學(xué)習(xí)學(xué)習(xí)吧2023-06-06
quartz的簡(jiǎn)單使用、SpringBoot使用和自定義數(shù)據(jù)源集成方式
這篇文章主要介紹了quartz的簡(jiǎn)單使用、SpringBoot使用和自定義數(shù)據(jù)源集成方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教<BR>2024-01-01
在日常開發(fā)中,IntelliJ?IDEA會(huì)通過(guò)problems窗口和編輯窗口的黃色標(biāo)記提示警告,這些警告可能指示代碼存在潛在風(fēng)險(xiǎn)或需要優(yōu)化的空間,文章介紹了如何利用IDEA檢查代碼,以及針對(duì)常見警告的原因和解決辦法2024-10-10