springboot整合RabbitMQ中死信隊列的實現(xiàn)
更新時間：2024年10月23日 08:36:45 作者：饅頭警告
死信是無法被消費的消息,產生原因包括消息TTL過期、隊列最大長度達到以及消息被拒絕且不重新排隊,RabbitMQ的死信隊列機制能夠有效防止消息數(shù)據(jù)丟失,適用于訂單業(yè)務等場景,本文就來介紹一下
一、死信的概念
先從概念解釋上搞清楚這個定義，死信，顧名思義就是無法被消費的消息，字面意思可以這樣理
解，一般來說，producer 將消息投遞到 broker 或者直接到 queue 里了，consumer 從 queue 取出消息進行消費，但某些時候由于特定的原因導致 queue 中的某些消息無法被消費，這樣的消息如果沒有后續(xù)的處理，就變成了死信，有死信自然就有了死信隊列。
應用場景:為了保證訂單業(yè)務的消息數(shù)據(jù)不丟失，需要使用到 RabbitMQ 的死信隊列機制，當消息
消費發(fā)生異常時，將消息投入死信隊列中.還有比如說: 用戶在商城下單成功并點擊去支付后在指定時間未支付時自動失效
二、死信的來源
1、消息 TTL 過期
2、隊列達到最大長度(隊列滿了，無法再添加數(shù)據(jù)到 mq 中)
3、消息被拒絕(basic.reject 或 basic.nack)并且 requeue=false.
三、死信實戰(zhàn)
1、代碼架構圖
生產者正常情況下走的是普通的交換機，這個交換機的類型是 direct ，它和普通隊列之間的關系是一個叫 "zhangsan" 的路由 key， 正常情況下會被 C1 消費。
但是發(fā)生了上面所說的三種情況中的一種，成為了死信，然后被轉換到死信交換機中，這個死信交換機也是 direct 類型，它們之間的 routingKey 是 "lisi",然后就進入了死信隊列，死信隊列由 C2 消費。
2、消息 TTL 過期
（1）消費者
// 死信隊列 實戰(zhàn)
// 消費者 1
public class Comsumer01 {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過期時間
//arguments.put("x-message-ttl",100000);
// 正常隊列設置死信交換機
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機與普通隊列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機與死信的隊列
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);
}
}// 消費者 2
public class Comsumer02 {
// 死信隊列名稱
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）生產者
// 死信隊列 生產者代碼
public class Producer {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設置 TTL 的時間
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）結果展示
3、隊列達到最大長度
（1）消費者
// 死信隊列 實戰(zhàn)
// 消費者 1
public class Comsumer01 {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過期時間
//arguments.put("x-message-ttl",100000);
// 正常隊列設置死信交換機
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
// 設置正常隊列的長度的限制
arguments.put("x-max-length",6);
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機與普通隊列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機與死信的隊列
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);
}
}// 死信隊列 實戰(zhàn)
// 消費者 2
public class Comsumer02 {
// 死信隊列名稱
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）生產者
// 死信隊列 生產者代碼
public class Producer {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設置 TTL 的時間
/*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）結果展示
4、消息被拒
（1）消費者
// 死信隊列 實戰(zhàn)
// 消費者 1
public class Comsumer01 {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
// 死信交換機名稱
public static final String DEAD_EXCHANGE = "dead_exchange";
// 普通隊列名稱
public static final String NORMAL_QUEUE = "normal_queue";
// 死信隊列名稱
public static final String DEAD_QUEUE = "dead_queue";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 交換機的聲明
channel.exchangeDeclare(NORMAL_EXCHANGE, BuiltinExchangeType.DIRECT);
channel.exchangeDeclare(DEAD_EXCHANGE,BuiltinExchangeType.DIRECT);
// 普通隊列的聲明
Map<String, Object> arguments = new HashMap<>();
// 過期時間
//arguments.put("x-message-ttl",100000);
// 正常隊列設置死信交換機
arguments.put("x-dead-letter-exchange",DEAD_EXCHANGE);
// 設置死信 RoutingKey
arguments.put("x-dead-letter-routing-key","lisi");
// 設置正常隊列的長度的限制
// arguments.put("x-max-length",6);
channel.queueDeclare(NORMAL_QUEUE,false,false,false,arguments);
// 死信隊列的聲明
channel.queueDeclare(DEAD_QUEUE,false,false,false,null);
// 綁定普通的交換機與普通隊列
channel.queueBind(NORMAL_QUEUE,NORMAL_EXCHANGE,"zhangsan");
// 綁定死信的交換機與死信的隊列
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 拒絕的");
// 拒絕，且不放囧普通隊列
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->{};
// 開啟手動應答
channel.basicConsume(NORMAL_QUEUE,false,deliverCallback,cancelCallback);
}
}// 死信隊列 實戰(zhàn)
// 消費者 2
public class Comsumer02 {
// 死信隊列名稱
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）生產者
// 死信隊列 生產者代碼
public class Producer {
// 普通交換機名稱
public static final String NORMAL_EXCHANGE = "normal_exchange";
public static void main(String[] args) throws IOException, TimeoutException {
Channel channel = RabbitMqUtils.getChannel();
// 死信消息 設置 TTL 的時間
/*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）結果展示
到此這篇關于springboot整合RabbitMQ中死信隊列的實現(xiàn)的文章就介紹到這了,更多相關springboot RabbitMQ死信隊列內容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
SpringBoot+websocket實現(xiàn)消息對話功能
WebSocket是一種在Web應用程序中實現(xiàn)實時雙向通信的技術,它可以用于在線游戲、在線聊天、推送通知、實時監(jiān)控等,并且比傳統(tǒng)的輪詢技術更加高效和可靠,本文就給大家介紹基于SpringBoot+websocket實現(xiàn)消息對話功能,感興趣的小伙伴可以自己動手試一試2023-09-09
這篇文章主要介紹了SpringBoot整合Log4j2以及配置詳解,刪除spring-boot-starter-parent默認使用spring-boot-starter-logging依賴，本文結合實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-01-01
這篇文章主要介紹了Spring Security基本架構與初始化操作流程，Spring Security是一個能夠為基于Spring的企業(yè)應用系統(tǒng)提供聲明式的安全訪問控制解決方案的安全框架2023-03-03
文章主要介紹了如何在Java項目中使用JWT進行用戶認證和授權,通過定義一個常量,編寫JWT工具類來生成和解析token,登錄時在服務端生成token并返回給客戶端,客戶端使用攔截器攔截請求,驗證token的有效性,從而實現(xiàn)權限控制,文章旨在分享個人經驗,為開發(fā)者提供參考2024-11-11
這篇文章主要介紹了Comet框架Plushlet使用問題解決方案,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2020-10-10
Elasticsearch倒排索引詳解及實際應用中的優(yōu)化
Elasticsearch(ES)使用倒排索引來加速文本的搜索速度,倒排索引之所以高效,主要是因為它改變了數(shù)據(jù)的組織方式,使得查詢操作可以快速完成,這篇文章主要給大家介紹了關于Elasticsearch倒排索引詳解及實際應用中優(yōu)化的相關資料,需要的朋友可以參考下2024-08-08
這篇文章主要為大家詳細介紹了如何使用Java實現(xiàn)將Word與PDF轉換為圖片,文中的示例代碼講解詳細,感興趣的小伙伴可以跟隨小編一起學習一下2024-10-10
這篇文章主要為大家詳細介紹了Java精確抽取網頁發(fā)布時間的相關資料，盡量做到精確無誤，感興趣的小伙伴們可以參考一下2016-06-06
在社交平臺上,熱搜功能是一個非常重要的組成部分,它展示了當前最熱門的話題,幫助用戶迅速了解最受關注的事件,Redis 是一個高性能的鍵值存儲系統(tǒng),通常用于緩存和實時數(shù)據(jù)存儲,本文將通過 Java 結合 Redis 實現(xiàn)一個簡化版的微博熱搜功能,需要的朋友可以參考下2024-12-12
SpringBoot接受前臺參數(shù)的6種方式以及統(tǒng)一響應代碼示例
這篇文章主要給大家介紹了關于SpringBoot接受前臺參數(shù)的6種方式以及統(tǒng)一響應的相關資料,前端負責展示頁面和用戶交互,而后端則負責處理業(yè)務邏輯和數(shù)據(jù)存儲,在這種架構下前端需要將用戶輸入的數(shù)據(jù)發(fā)送給后端進行處理,需要的朋友可以參考下2023-12-12