Java RabbitMQ的工作隊列與消息應答詳解
Work Queues
工作隊列（任務隊列）主要思想是避免立即執(zhí)行資源密集型任務，而不得不等待它完成，相反我們安排任務在之后執(zhí)行。我們把任務封裝為消息并將其發(fā)送到隊列。在后臺運行的工作進程將彈出任務并最終執(zhí)行作業(yè)。當有多個工作線程時，這些工作線程將一起處理這些任務。
其實就是生產(chǎn)者發(fā)送大量的消息，發(fā)送到隊列之后，由多個消費者(工作線程)來處理消息，并且每個消息只能被處理一次。
1. 輪詢分發(fā)消息
多個工作線程按照次序每來一個消息執(zhí)行一次。
1.1 抽取工具類
直接通過信息獲取信道
/**
* @Description RabbitMQ工具類
* @date 2022/3/5 10:02
*/
public class RabbitMQUtils {
public static Channel getChannel() throws Exception{
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("1");
factory.setUsername("guest");
factory.setPassword("guest");
Connection connection = factory.newConnection();
return connection.createChannel();
}
}
1.2 編寫兩個工作線程
Work2和Work1代碼沒有區(qū)別，只需要對它做出區(qū)分即可。
public class Worker1 {
// 指定隊列名稱
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
// 獲取信道
Channel channel = RabbitMQUtils.getChannel();
// 聲明：接收消息回調(diào)
DeliverCallback deliverCallback = (consumerTag, message) -> {
System.out.println("工作線程01："+ new String(message.getBody()));
};
// 聲明：取消消費回調(diào)
CancelCallback cancelCallback = consumerTag -> {
System.out.println("工作線程01取消接收："+consumerTag);
};
System.out.println("工作線程01啟動完成......");
channel.basicConsume(QUEUE_NAME,false,deliverCallback,cancelCallback);
}
}
1.3 編寫生產(chǎn)者
public class Producer {
private static final String QUEUE_NAME = "work_queue";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQUtils.getChannel();
// 產(chǎn)生隊列
channel.queueDeclare(QUEUE_NAME,false,false,true,null);
// 消息體
Scanner scanner = new Scanner(System.in);
int i = 1;
while (scanner.hasNext()){
String msg = scanner.next();
msg = msg + i;
channel.basicPublish("",QUEUE_NAME,null,msg.getBytes());
System.out.println("發(fā)送成功：" + msg);
}
System.out.println("----------==========發(fā)送完畢==========----------");
}
}
1.4 運行測試
先啟動兩個工作線程，再啟動生產(chǎn)者。
出現(xiàn)404異常請參考下方1.6
生產(chǎn)者發(fā)送情況：
輪詢狀態(tài)下兩個工作隊列接收狀態(tài)：
1.5 異常情況
在先啟動兩個消費者線程時，會提示404找不到隊列。
Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method<channel.close>(reply-code=404, reply-text=NOT_FOUND - no queue 'work_queue' in vhost '/', class-id=60, method-id=20)
發(fā)生這個情況的原因很顯然是因為先啟動了消費者，但是在RabbitMQ中沒有創(chuàng)建相對應的隊列名稱，解決方法可以：
1.先啟動生產(chǎn)者創(chuàng)建隊列（也可以在RabbitMQ中創(chuàng)建隊列）；
2.再啟動消費者就不會產(chǎn)生這個錯誤；
3.再在生產(chǎn)者中使用Scanner類去發(fā)送消息測試。
2. 消息應答
消費者在接收到消息并且處理該消息之后，告訴RabbitMQ它已經(jīng)處理了，RabbitMQ可以刪除消息。其目的就是為了保護消息在被處理之前不會消失。
2.1 自動應答
這種方式發(fā)送后就被認定為已經(jīng)傳送成功，所以在消息接收到之前消費者的連接或者channel關閉，那么這個消息就會丟失。其特點是消費者可以傳遞過載的消息，對傳遞的消息沒有限制，但如果因內(nèi)存耗盡消費者線程被系統(tǒng)殺死，就會使得多條消息丟失。所以這個模式需要在數(shù)據(jù)安全性和吞吐量之間選擇，適合使用在消費者可以高效并以某種速率能夠處理這些消息的情況下使用。
所以自動應答的方式局限性很高。
2.2 手動應答
優(yōu)點：可以批量應答和減少網(wǎng)絡擁擠。
1.channel.basicAck(long deliveryTag, boolean multiple);：肯應應答，處理完消息之后提醒RabbitMQ可以刪除當前隊列，deliveryTag：當前隊列中選中的消息；multiple：是否批量應答。
2.channel.basicNack(long deliveryTag, boolean multiple, boolean requeue)：否定應答，
3.channel.basicReject(long deliveryTag, boolean requeue)：否定并且拒絕應答。
2.3 消息自動重新入隊
如果消費者因為一些原因失去了對RabbitMQ的連接，導致沒有發(fā)送ACK確認，RabbitMQ就會對該消息進行重新排隊，并且分發(fā)給可以處理該消息的消費者，所以即使某個消費者死亡，也可以保證消息不會丟失。
2.4 手動應答測試
測試目的：在手動應答狀態(tài)下不會發(fā)生消息丟失的情況。
測試方法：
1.創(chuàng)建兩個消費者；
2.使用工具類使線程睡眠一定時間；
3.在睡眠時關閉線程，看能否自動重新入隊。
2.4.1 生產(chǎn)者代碼
/**
* @Description 手動應答生產(chǎn)者
* @date 2022/3/5 19:03
*/
public class Producer1 {
// 指定隊列名
private static final String TASK_QUEUE_RES = "queue_res";
public static void main(String[] args) throws Exception {
Channel channel = RabbitMQUtils.getChannel();
channel.queueDeclare(TASK_QUEUE_RES,false,false,false,null);
Scanner scanner = new Scanner(System.in);
int i = 0;
while (scanner.hasNext()){
i++;
String msg = scanner.next() + i;
channel.basicPublish("",TASK_QUEUE_RES,null,msg.getBytes(StandardCharsets.UTF_8));
System.out.println("發(fā)送消息：'" + msg + "'成功");
}
}
}
2.4.2 消費者代碼
/**
* @Description 手動應答消費者1
* @date 2022/3/5 19:17
*/
public class Worker1 {
private static final String TASK_QUEUE_RES = "queue_res";
public static void main(String[] args) throws Exception{
Channel channel = RabbitMQUtils.getChannel();
System.out.println("線程A等待接收......");
DeliverCallback deliverCallback = (consumerTag, message) -> {
// 模擬并發(fā)沉睡一秒
try {
Thread.sleep(1000);
System.out.println("線程A接收消息："+ new String(message.getBody(), StandardCharsets.UTF_8));
/**
* basicAck:
* 1. 消息標記
* 2. 是否批量
*/
channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
} catch (InterruptedException e) {
e.printStackTrace();
}
};
channel.basicConsume(TASK_QUEUE_RES,false,deliverCallback,
consumerTag -> {
System.out.println(consumerTag + "消費者取消消費");
});
}
}
Worker2類和1區(qū)別不大，將名稱改成B再將睡眠事件改成30即可。
2.4.3 測試
測試方法：
1.先啟動生產(chǎn)者創(chuàng)建隊列；
2.啟動兩個消費者接收消息；
3.因為是輪詢方式，所以A線程接收之后肯定是B線程接收，在睡眠時關閉B線程，如果A線程接收到說明測試成功。
發(fā)送消息：
線程A接收：
再發(fā)送消息：
關閉線程B線程A接收到消息：
測試成功！
總結
本篇文章就到這里了，希望能夠給你帶來幫助，也希望您能夠多多關注腳本之家的更多內(nèi)容!
相關文章
Apache Commons Math3探索之快速傅立葉變換代碼示例
這篇文章主要介紹了Apache Commons Math3探索之快速傅立葉變換代碼示例，具有一定參考價值，需要的朋友可以了解下。2017-10-10
Java并發(fā)編程Semaphore計數(shù)信號量詳解
這篇文章主要介紹了Java并發(fā)編程Semaphore計數(shù)信號量詳解，具有一定參考價值，需要的朋友可以了解下。2017-10-10
這篇文章主要介紹了基于Java?GUI?事件處理方式，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-12-12
這篇文章主要介紹了如何避免在Java中使用雙括號初始化,本文通過實例代碼給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2023-07-07
這篇文章主要介紹了解決SpringMVC攔截器path路徑的坑，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-10-10
這篇文章主要介紹了java異常處理攔截器，使用異常處理攔截器，可以不用寫那么多try…catch…，下面就來學習關于java異常處理攔截器的詳情內(nèi)容吧,需要的朋友可以參考一下2021-10-10
這篇文章主要介紹了Java的Map中computeIfAbsent方法詳解,在jdk1.8中Map接口新增了一個computeIfAbsent方法,這是Map接口中的默認實現(xiàn)該方法是首先判斷緩存Map中是否存在指定的key的值,如果不存在,會調(diào)用mappingFunction(key)計算key的value,需要的朋友可以參考下2023-11-11
說起內(nèi)部類這個詞，想必很多人都不陌生，但是又會覺得不熟悉。原因是平時編寫代碼時可能用到的場景不多，用得最多的是在有事件監(jiān)聽的情況下，并且即使用到也很少去總結內(nèi)部類的用法。今天我們就來一探究竟2022-04-04
這篇文章主要介紹了分別通過Java和C++實現(xiàn)滑動窗口最大值，即給定一個數(shù)組?nums?和滑動窗口的大小?k，請找出所有滑動窗口里的最大值。感興趣的可以了解一下2021-12-12
這篇文章主要為大家詳細介紹了java設計模式之中介者模式，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2016-08-08