前面幾篇案例已經將常用的交換器（DirectExchange、TopicExchange、FanoutExchange）的用法介紹完了，現在我們來看一下消息的回調，也就是消息確認。
在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之消息確認的文章就介紹到這了,更多相關SpringBoot整合RabbitMQ消息確認內容請搜索服務器之家以前的文章或繼續瀏覽下面的相關文章希望大家以后多多支持服務器之家！
原文鏈接：https://www.cnblogs.com/liweixml/p/14662355.html
