RabbitMQ消息確認機制剖析
前言
上一章講解了RabbitMq的三種Exchange消息發(fā)送的模式，但是在默認情況下RabbitMQ并不能保證消息是否發(fā)送成功，以及是否能被成功消費，為了保證消息在傳遞過程中不丟失，需要對消息進行確認機制，來提高消息的可靠性。
消息確認
基本流程
說明：
- 生產者發(fā)送消息到RabbitMQ Server后，RabbitMQ Server需要對生產者進行消息Confirm確認。
- 消費者消費消息后需要對 RabbitMQ Server進行消息ACK確認。
消息確認模式
RabbitMq提供了兩種消息發(fā)送者確認模式分別為： ConfirmCallback確認模式和 ReturnCallback退回模式。
ConfirmCallback確認模式
@Component
public class RabbitConfirmConfig implements ConfirmCallback
{
private Logger logger = LoggerFactory.getLogger(RabbitConfirmConfig.class);
public void confirm(CorrelationData correlationData, boolean ack,
String cause)
{
logger.info("數(shù)據(jù)內容:{}",correlationData);
logger.info("是否確認成功:{}",ack);
logger.info("錯誤原因:{}",cause);
if (!ack)
{
logger.info("exchange produce confirm message send error" + cause);
}
else
{
logger.info("exchange produce confirm message send success");
}
}
}
說明：ConfirmCallback模式確認，需要重寫confirm接方法，此方法的三個參數(shù)分別為：CorrelationData、ack、cause
- CorrelationData:對象內部只有一個id屬性，用來表示當前消息的唯一性。
- ack:消息投遞狀態(tài)，true表示投遞成功
- cause: 消息投遞失敗原因
雖然消息被broker接收到只能表示已經到達MQ服務器，但是并不能保證消息一定會被投遞到目標 queue里。所以我們需要實現(xiàn)returnCallback來進行相關處理。
ReturnCallback退回模式
@Component
public class RabbitReturnConfig implements ReturnCallback
{
private Logger logger = LoggerFactory.getLogger(RabbitReturnConfig.class);
public void returnedMessage(Message message, int replyCode,
String replyText, String exchange, String routingKey)
{
logger.info("消息發(fā)送送到隊列信息：");
logger.info("發(fā)生消息：{}",message);
logger.info("回應碼：{}",replyCode);
logger.info("回應信息：{}",replyText);
logger.info("交換機：{}",exchange);
logger.info("路由鍵：{}",routingKey);
}
}
說明：實現(xiàn)接口ReturnCallback重寫returnedMessage()方法，方法有五個參數(shù)message（消息體）、replyCode（響應code）、replyText（響應內容）、exchange（交換機）、routingKey（路由鍵）。
消息發(fā)送者確認
@Component
public class MqConfirmProduce
{
@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private RabbitConfirmConfig rabbitConfirmConfig;
@Autowired
private RabbitReturnConfig rabbitReturnConfig;
/**
*
* @param exchange 消息交互機名稱
* @param routeKey 消息路由鍵的名稱
* @param message 消息內容
*/
public void sendMessage(String exchange ,String routeKey,Object msg)
{
//確保消息發(fā)送失敗后可以重新返回到隊列中
rabbitTemplate.setMandatory(true);
// 消費者確認收到消息后，手動ack回執(zhí)回調處理
rabbitTemplate.setConfirmCallback(rabbitConfirmConfig);
//消息投遞到隊列失敗回調處理
rabbitTemplate.setReturnCallback(rabbitReturnConfig);
//保證消息唯一性
CorrelationData correlationData =new CorrelationData(UUID.randomUUID().toString());
//發(fā)送消息
rabbitTemplate.convertAndSend(exchange,routeKey,msg,
message -> {
message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
return message;
},
correlationData);
}
}
說明：注意需要開啟消息確認的配置：
rabbitmq:
host: localhost
port: 5672
username: guest
password: guest
virtual-host: /
#開啟發(fā)送確認
publisher-confirms: true
# 開啟發(fā)送失敗退回
publisher-returns: true
listener:
simple:
# 手動確認
acknowledge-mode: manual
retry:
enabled: true
消息接收者確認
@Component
@RabbitListener(queues = "testQueue")
public class MqConfirmConsumer
{
private static final Logger logger = LoggerFactory.getLogger(MqConfirmConsumer.class);
@RabbitHandler
public void receive(String msg, Channel channel, Message message) throws IOException
{
logger.info("receive message content:{}",message);
try
{
logger.info("開始消息確認");
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
logger.info("消息確認成功");
}
catch (Exception e)
{
logger.error("消息確認失敗，即將再次返回隊列中"); channel.basicNack(message.getMessageProperties().getDeliveryTag(), true, true);
}
}
}
說明：消息者確認消息有三種模式，分別為basicAck、basicNack、basicReject。
basicAck模式
表示成功確認，使用此回執(zhí)方法后，消息會被rabbitmq broker刪除。
void basicAck(long deliveryTag, boolean multiple)
- deliveryTag：消息投遞序號，
- multiple：是否批量確認，值為 true則會一次性ack所有小于當前消息deliveryTag的消息。
basicNack模式
表示失敗確認，一般在消費消息異常時用到此方法，可以將消息重新投遞入隊列。
void basicNack(long deliveryTag, boolean multiple, boolean requeue)
- deliveryTag：表示消息投遞序號。
- requeue： 表示消息是否重新入隊列，true表示重新投入隊列中。
- multiple：是否批量確認，true表示會一次性ack所有小于當前消息deliveryTag的消息。
basicReject模式
basicReject：拒絕消息，與basicNack區(qū)別在于不能進行批量操作，其他用法很相似。
void basicReject(long deliveryTag, boolean requeue)
- deliveryTag：消息投遞序號。
- requeue：值為true表示消息重新入隊列
測試
測試發(fā)送消息，消息發(fā)送者的確認信息如下：
c.s.f.r.config.RabbitConfirmConfig - exchange produce confirm message send success
c.s.f.r.config.RabbitConfirmConfig - 數(shù)據(jù)內容:CorrelationData [id=88ea47a5-726d-44c5-9839-1f2a6bf942ed]
c.s.f.r.config.RabbitConfirmConfig - 是否確認成功:true
c.s.f.r.config.RabbitConfirmConfig - 錯誤原因:null
c.s.f.r.config.RabbitConfirmConfig - exchange produce confirm message send success
消費者的確認信息如下：
receive message content:(Body:'this is test message' MessageProperties [headers={spring_listener_return_correlation=0fcefb6d-acea-4eb2-8484-e3a82f8c584f, spring_returned_message_correlation=88ea47a5-726d-44c5-9839-1f2a6bf942ed}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=testDirect, receivedRoutingKey=testDirectRouting, deliveryTag=2, consumerTag=amq.ctag-dOwkSPuI1e0HR_1Ufu3Erw, consumerQueue=testQueue])
c.s.f.r.consumer.MqConfirmConsumer - 開始消息確認
c.s.f.r.consumer.MqConfirmConsumer - 消息確認成功
消費者確認失敗
如果消息確認在消費者確認失敗，那么消息將會重寫投遞導導消息隊列的首部。模擬消費者確認失敗場景：
@Component
@RabbitListener(queues = "testQueue")
public class MqConfirmConsumer
{
private static final Logger logger = LoggerFactory.getLogger(MqConfirmConsumer.class);
@RabbitHandler
public void receive(String msg, Channel channel, Message message) throws IOException
{
logger.info("receive message content:{}",message);
try
{
logger.info("開始消息確認");
int c=1/0;
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
logger.info("消息確認成功");
}
catch (Exception e)
{
logger.error("消息確認失敗，即將再次返回隊列中"); channel.basicNack(message.getMessageProperties().getDeliveryTag(), true, true);
}
}
}
查看執(zhí)行結果：
c.s.f.r.consumer.MqConfirmConsumer - receive message content:(Body:'this is test message' MessageProperties [headers={spring_listener_return_correlation=0fcefb6d-acea-4eb2-8484-e3a82f8c584f, spring_returned_message_correlation=39d4cdd1-cbeb-4090-91ea-9e5d0bed785c}, contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT, priority=0, redelivered=false, receivedExchange=testDirect, receivedRoutingKey=testDirectRouting, deliveryTag=1, consumerTag=amq.ctag-e5GtG455pkm7eWfY3xGleg, consumerQueue=testQueue])
c.s.f.r.consumer.MqConfirmConsumer - 開始消息確認
c.s.f.r.consumer.MqConfirmConsumer - 消息確認失敗，即將再次返回隊列中
消息已經重新返回隊列中。我們查看隊列信息具體如下：
說明：我們可以看到消息為Unacked狀態(tài)，消息又會重新會被消費，然后確認失敗，又重新被消費，導致死循環(huán)。
解決辦法
針對這種情況，我們將如何處理呢？我們手動確認失敗后，并將消息持久入到MySQL中通過定時任務做補償。然后刪除消息隊列。具體修改如下：
@RabbitHandler
public void receive(String msg, Channel channel, Message message) throws IOException
{
logger.info("receive message content:{}",message);
try
{
logger.info("開始消息確認");
int c=1/0;
channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
logger.info("消息確認成功");
}
catch (Exception e)
{
if (message.getMessageProperties().getRedelivered())
{
logger.error("消息確認失敗,拒絕處理");
//執(zhí)行持久化處理 channel.basicReject(message.getMessageProperties().getDeliveryTag(), false);
}
else
{
logger.error("消息確認失敗，即將再次返回隊列中");
channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
}
}
}
修改后執(zhí)行結果如下：
總結
本文講解了RabbitMQ消息確認機制，消息是否需要確認，我們需要根據(jù)業(yè)務的場景來分析，如有疑問，請隨時反饋，更多關于RabbitMQ消息確認的資料請關注腳本之家其它相關文章！
相關文章
springboot整合RabbitMQ發(fā)送短信的實現(xiàn)
本文會和SpringBoot做整合，實現(xiàn)RabbitMQ發(fā)送短信，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2021-05-05
詳解mall整合SpringBoot+MyBatis搭建基本骨架
這篇文章主要介紹了詳解mall整合SpringBoot+MyBatis搭建基本骨架，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-08-08
Java編程一維數(shù)組轉換成二維數(shù)組實例代碼
這篇文章主要介紹了Java編程一維數(shù)組轉換成二維數(shù)組，分享了相關代碼示例，小編覺得還是挺不錯的，具有一定借鑒價值，需要的朋友可以參考下2018-01-01
@Import注解算是SpringBoot自動配置原理中一個很重要的注解，本文介紹了該注解的源碼分析及使用方法，感興趣的朋友可以了解下2021-05-05
本篇文章主要介紹了datatables 帶查詢條件java服務端分頁處理實例，具有一定的參考價值，感興趣的小伙伴們可以參考一下2017-06-06
下面小編就為大家?guī)硪黄赗abbitMQ幾種Exchange 模式詳解。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-11-11
計算器是我們日常生活中常用的工具之一,它能夠進行基本的數(shù)學運算,如加法、減法、乘法和除法,而在設計一個計算器時,我們可以通過使用Java編程語言來實現(xiàn)一個簡單的控制臺計算器,并且讓它能夠保留計算過程,文中有詳細的代碼示例,需要的朋友可以參考下2023-11-11
本文為大家提供一個java向文件末尾添加內容的示例分享，大家參考使用吧2014-01-01
對dbunit進行mybatis DAO層Excel單元測試(必看篇)
下面小編就為大家?guī)硪黄獙bunit進行mybatis DAO層Excel單元測試(必看篇)。小編覺得挺不錯的，現(xiàn)在就分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2017-05-05
本文主要介紹了快速搭建springboot項目(新手入門)，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2023-07-07