本篇內容介紹了“利用SpringBoot整合RabbitMQ實現消息確認機制”的有關知識，在實際案例的操作過程中，不少人都會遇到這樣的困境，接下來就讓小編帶領大家學習一下如何處理這些情況吧！希望大家仔細閱讀，能夠學有所成！
在rabbitmq-provider項目的application.yml文件上加上一些配置
server: port: 8021 spring: #給項目來個名字 application: name: rabbitmq-provider #配置rabbitMq 服務器 rabbitmq: host: 127.0.0.1 port: 5672 username: guest password: guest # connection-timeout: 60s #虛擬host 可以不設置,使用server默認host #virtual-host: admin # 確認消息已發送到交換機（Exchange） publisher-confirm-type: correlated # 確認消息已發送到隊列 publisher-returns: true
創建RabbitConfig配置類，配置消息的相關回調函數
package com.rabbitmq.provider.config;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class RabbitConfig {
@Bean
DirectExchange lonelyDirectExchange() {
return new DirectExchange("lonelyDirectExchange");
}
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory){
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
// 設置開啟Mandatory，才能觸發回調函數，無論消息推送結果怎么樣都強制調用回調函數
rabbitTemplate.setMandatory(true);
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean b, String s) {
System.out.println("ConfirmCallback ：相關數據：" + correlationData);
System.out.println("ConfirmCallback ：確認情況：" + b);
System.out.println("ConfirmCallback ：原因：" + s);
}
});
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int i, String s, String s1, String s2) {
System.out.println("ReturnCallback ：消息：" + message);
System.out.println("ReturnCallback ：回應碼：" + i);
System.out.println("ReturnCallback ：回應消息：" + s);
System.out.println("ReturnCallback ：交換機：" + s1);
System.out.println("ReturnCallback ：路由鍵：" + s2);
}
});
return rabbitTemplate;
}
}生產者回調函數已經配置完畢，上面我們配置了兩個回調函數setConfirmCallback和setReturnCallback
一般會產生四種情況：
①消息推送到server，但是在server里找不到交換機
②消息推送到server，找到交換機了，但是沒找到隊列
③消息推送到sever，交換機和隊列啥都沒找到
④消息推送成功
接下來分別測試和認證下以上4種情況，消息確認觸發回調函數的情況：
①消息推送到server，但是在server里找不到交換機
寫個測試接口，把消息推送到名為‘non-existent-exchange'的交換機上（這個交換機是沒有創建沒有配置的）：
@GetMapping("/testMessageAck")
public String testMessageAck() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: non-existent-exchange test message ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> map = new HashMap<>();
map.put("messageId", messageId);
map.put("messageData", messageData);
map.put("createTime", createTime);
rabbitTemplate.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
return "ok";
}調用接口，查看rabbitmq-provuder項目的控制臺輸出情況（原因里面有說，沒有找到交換機'non-existent-exchange'）：
所以證明在交換器沒有配置，推送消息找不到對應的交換器，會執行ConfirmCallback方法。
②消息推送到server，找到交換機了，但是沒找到隊列
我們在RabbitConfig里面配置了一個名為lonelyDirectExchange的交換器，再新增一個接口testMessageAck2，使用lonelyDirectExchange交換器
@GetMapping("/testMessageAck2")
public String testMessageAck2() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: non-existent-exchange test message ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> map = new HashMap<>();
map.put("messageId", messageId);
map.put("messageData", messageData);
map.put("createTime", createTime);
rabbitTemplate.convertAndSend("lonelyDirectExchange", "TestDirectRouting", map);
return "ok";
}重啟服務，使用postman調用這個接口
可以看到ReturnCallback和ConfirmCallback這兩個都被調用了。
CallfirmCallback的確認情況為true，因為消息是推送成功到服務器了的，所以ConfirmCallback對消息確認情況是true
ReturnCallback的回應消息為NO_ROUTE找不到隊列
③消息推送到sever，交換機和隊列啥都沒找到
這種情況和1、2很相似，會調用ReturnCallback和CallfirmCallback這兩個回調函數
④消息推送成功
我們調用之前的sendFanoutMessage接口來看一下會是什么情況
結論：消息推送成功調用ConfirmCallback回調函數
消息接收確認機制主要存在三種模式：
①：自動確認
②：根據情況確認
③：手動確認
接下來，我們看看手動確認是怎么實現的
在rabbitmq-consumer消費者項目上創建MessageListenerConfig配置類
package com.rabbitmq.consumer.config;
import com.rabbitmq.consumer.receiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@Configuration
public class MessageListenerConfig {
@Autowired
private CachingConnectionFactory connectionFactory;
@Autowired
private MyAckReceiver myAckReceiver; //消息接收處理類
@Bean
public SimpleMessageListenerContainer simpleMessageListenerContainer() {
SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
container.setConcurrentConsumers(1);
container.setMaxConcurrentConsumers(1);
// RabbitMQ 默認是自動確認，改成手動確認
container.setAcknowledgeMode(AcknowledgeMode.MANUAL);
// 設置隊列
// 如果同時設置多個如下:前提是隊列是必須已經創建存在的
// container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
// 設置一個隊列
// container.setQueues(new Queue("TestDirectQueue",true));
// 設置多個隊列
container.addQueues(new Queue("TestDirectQueue",true));
container.addQueues(new Queue("fanout.A",true));
// container.addQueues(new Queue("TestDirectQueue3",true));
container.setMessageListener(myAckReceiver);
return container;
}
}創建手動確定消息監聽類MyAckReceiver
package com.rabbitmq.consumer.receiver;
import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;
import java.util.HashMap;
import java.util.Map;
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//因為傳遞消息的時候用的map傳遞，所以將map從Message內取出來需要做些處理
String msg = message.toString();
String[] msgArray = msg.split("'");
Map<String,String> msgMap = mapStringToMap(msgArray[1].trim(),3);
String messageId = msgMap.get("messageId");
String messageData = msgMap.get("messageData");
String createTime = msgMap.get("createTime");
System.out.println("MyAckReceiver ：");
System.out.println("messageId = " + messageId);
System.out.println("messageData = " + messageData);
System.out.println("createTime = " + createTime);
System.out.println("消費的主題來自：" + message.getMessageProperties().getConsumerQueue());
if("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
System.out.println("執行TestDirectQueue中的消息的業務處理流程.....");
}
if("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
System.out.println("執行fanout.A中的消息的業務處理流程.....");
}
//第二個參數，手動確認可以被批處理，當該參數為 true 時，則可以一次性確認 delivery_tag 小于等于傳入值的所有消息
channel.basicAck(deliveryTag,true);
//第二個參數，true會重新放回隊列，所以需要自己根據業務邏輯判斷什么時候使用拒絕
//channel.basicReject(deliveryTag, true);
}catch (Exception e){
channel.basicAck(deliveryTag,false);
e.printStackTrace();
}
}
// {key=value,key=value,key=value} 格式轉換成map
private Map<String,String> mapStringToMap(String str,int entryNum) {
str = str.substring(1, str.length()-1);
String[] strs = str.split(",", entryNum);
Map<String,String> map = new HashMap<String,String>();
for (String s : strs) {
String key = s.split("=")[0].trim();
String value = s.split("=")[1];
map.put(key,value);
}
return map;
}
}重啟服務，使用postman調用sendDirectMessage接口
再調用sendFanoutMessage接口
如果還想添加其它處處理器手動確認，按照上面的模式繼續添加即可。
“利用SpringBoot整合RabbitMQ實現消息確認機制”的內容就介紹到這里了，感謝大家的閱讀。如果想了解更多行業相關的知識可以關注億速云網站，小編將為大家輸出更多高質量的實用文章！
免責聲明：本站發布的內容（圖片、視頻和文字）以原創、轉載和分享為主，文章觀點不代表本網站立場，如果涉及侵權請聯系站長郵箱：is@yisu.com進行舉報，并提供相關證據，一經查實，將立刻刪除涉嫌侵權內容。
