RabbitMQ中Confirm消息確認(rèn)機(jī)制保障生產(chǎn)端消息的可靠性詳解
1. 概述
生產(chǎn)者將數(shù)據(jù)發(fā)送到 RabbitMQ 的時(shí)候,可能數(shù)據(jù)就在半路給搞丟了,因?yàn)榫W(wǎng)絡(luò)問題啥的,都 有可能。此時(shí)可以開啟 confirm 模式,在生產(chǎn)者那里設(shè)置開啟 confirm 模式之后,你每次寫的消息都會(huì)分配一個(gè)唯一的 id,然后如果寫入了 RabbitMQ 中,RabbitMQ 會(huì)給你回傳一個(gè) ack 消息,告訴你說這個(gè)消息 ok 了。如果RabbitMQ 沒能處理這個(gè)消息,會(huì)回調(diào)你的一個(gè) nack 接口,告訴你這個(gè)消息接收失敗,你可以重試。而且你可以結(jié)合這個(gè)機(jī)制自己在內(nèi)存里維護(hù)每個(gè)消息 id 的狀態(tài),如果超過一定時(shí)間還 沒接收到這個(gè)消息的回調(diào),那么你可以重發(fā)。

在實(shí)際項(xiàng)目中,可以利用這一機(jī)制保障消息的可靠性投遞,如果消息未發(fā)送成功,可以在監(jiān)聽事件中記錄日志、重新發(fā)送消息等操作。
2.原生API中開啟Confirm消息確認(rèn)機(jī)制
- 在生產(chǎn)者的channel上開啟確認(rèn)機(jī)制: channel.confirmSelect();
- 在channel上添加Confirm監(jiān)聽事件: channel.addConfirmListener(new ConfirmListener() ...
2.1 代碼演示
生產(chǎn)者代碼
監(jiān)聽事件的兩個(gè)方法:handleAck() 消息投遞成功后回調(diào),handleNack 消息未成功投遞回調(diào)
public static void main(String[] args) throws Exception{
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//設(shè)置虛擬主機(jī)
connectionFactory.setVirtualHost("/");
//創(chuàng)建一個(gè)鏈接
Connection connection = connectionFactory.newConnection();
//創(chuàng)建channel
Channel channel = connection.createChannel();
//消息的確認(rèn)模式
channel.confirmSelect();
String exchangeName="test_confirm_exchange";
String routeKey="confirm.test";
String msg="RabbitMQ send message confirm test!";
for (int i=0;i<5;i++){
channel.basicPublish(exchangeName,routeKey,null,msg.getBytes());
}
//確定監(jiān)聽事件
channel.addConfirmListener(new ConfirmListener() {
/**
* 消息成功發(fā)送
* @param deliveryTag 消息唯一標(biāo)簽
* @param multiple 是否批量
* @throws IOException
*/
@Override
public void handleAck(long deliveryTag, boolean multiple) throws IOException {
System.out.println("**********Ack*********");
}
/**
* 消息沒有成功發(fā)送
* @param deliveryTag
* @param multiple
* @throws IOException
*/
@Override
public void handleNack(long deliveryTag, boolean multiple) throws IOException {
System.out.println("**********No Ack*********");
}
});
}消費(fèi)者端代碼
public static void main(String[] args) throws Exception{
System.out.println("======消息接收start==========");
ConnectionFactory connectionFactory=new ConnectionFactory();
connectionFactory.setHost("127.0.0.1");
connectionFactory.setPort(5672);
connectionFactory.setUsername("guest");
connectionFactory.setPassword("guest");
//設(shè)置虛擬主機(jī)
connectionFactory.setVirtualHost("/");
//創(chuàng)建鏈接
Connection connection = connectionFactory.newConnection();
//創(chuàng)建channel
Channel channel = connection.createChannel();
String exchangeName="test_confirm_exchange";
String exchangeType="topic";
//聲明Exchange
channel.exchangeDeclare(exchangeName,exchangeType,true,false,false,null);
String queueName="test_confirm_queue";
//聲明隊(duì)列
channel.queueDeclare(queueName,true,false,false,null);
String routeKey="confirm.#";
//綁定隊(duì)列和交換機(jī)
channel.queueBind(queueName,exchangeName,routeKey);
channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
System.out.println("接收到消息::"+new String(body));
}
});
}到此這篇關(guān)于RabbitMQ中Confirm消息確認(rèn)機(jī)制保障生產(chǎn)端消息的可靠性詳解的文章就介紹到這了,更多相關(guān)Confirm消息確認(rèn)機(jī)制內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
Spring?Boot面試必問之啟動(dòng)流程知識(shí)點(diǎn)詳解
SpringBoot是Spring開源組織下的子項(xiàng)目,是Spring組件一站式解決方案,主要是簡化了使用Spring的難度,簡省了繁重的配置,提供了各種啟動(dòng)器,開發(fā)者能快速上手,這篇文章主要給大家介紹了關(guān)于Spring?Boot面試必問之啟動(dòng)流程知識(shí)點(diǎn)的相關(guān)資料,需要的朋友可以參考下2022-06-06
Spring Boot 2和Redis例子實(shí)現(xiàn)過程解析
這篇文章主要介紹了Spring Boot2發(fā)布與調(diào)用REST服務(wù)過程解析,文中通過示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11
MyBatis實(shí)戰(zhàn)之動(dòng)態(tài)生成SQL詳解
這篇文章主要為大家詳細(xì)介紹了如何使用MyBatis實(shí)現(xiàn)動(dòng)態(tài)生成SQL,可以告別硬編碼,擁抱智能SQL生成,感興趣的小伙伴可以跟隨小編一起學(xué)習(xí)一下2025-07-07
redisson特性及優(yōu)雅實(shí)現(xiàn)示例
這篇文章主要為大家介紹了redisson特性及優(yōu)雅實(shí)現(xiàn)示例詳解,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進(jìn)步,早日升職加薪2022-11-11
Mybatis與Jpa的區(qū)別和性能對(duì)比總結(jié)
mybatis和jpa兩個(gè)持久層框架,從底層到用法都不同,但是實(shí)現(xiàn)的功能是一樣的,所以說一直以來頗有爭議,所以下面這篇文章主要給大家介紹了關(guān)于Mybatis與Jpa的區(qū)別和性能對(duì)比的相關(guān)資料,需要的朋友可以參考下2021-06-06
Java詳解HashMap實(shí)現(xiàn)原理和源碼分析
這篇文章主要介紹了Java關(guān)于HashMap的實(shí)現(xiàn)原理并進(jìn)行源碼分析,本文給大家介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2021-09-09

