RabbitMQ交換機與Springboot整合的簡單實現(xiàn)
RabbitMQ-交換機
1、交換機是干什么的？
消息(Message)由Client發(fā)送，RabbitMQ接收到消息之后通過交換機轉發(fā)到對應的隊列上面。Worker會從隊列中獲取未被讀取的數(shù)據(jù)處理。
1、交換機的種類
RabbitMQ包含四種不同的交換機類型：
- Direct exchange：直連交換機，轉發(fā)消息到routigKey指定的隊列
- Fanout exchange：扇形交換機，轉發(fā)消息到所有綁定隊列（速度最快）
- Topic exchange：主題交換機，按規(guī)則轉發(fā)消息（最靈活）
- Headers exchange：首部交換機 （不常用）
1、Direct exchange（直連交換機）
直連交換機是一種帶路由功能的交換機，根據(jù)消息攜帶的路由鍵將消息投遞給對應隊列。
一個隊列通過routing_key（路由鍵）與一個交換機綁定，當消息被發(fā)送的時候，需要指定一個routing_key，這個消息被送達交換機的時候，就會被交換機送到指定的隊列里面去。
類似一對一的關系！
1.1 Springboot的簡單實現(xiàn)
創(chuàng)建2個springboot項目，一個 rabbitmq-provider （生產(chǎn)者），一個rabbitmq-consumer（消費者）。
1、首先創(chuàng)建 rabbitmq-provider
pom.xml添加依賴：
<!--rabbitmq-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-amqp</artifactId>
</dependency>
配置application.yml：
server:
port: 8090
spring:
application:
name: rabbitmq-provider
rabbitmq:
host: 192.168.152.173
port: 5672 #注意15672是web頁面端口，消息發(fā)送端口為5672
username: guest
password: guest
編寫配置文件
@Configuration
public class DirectRabbitConfig {
/**
* 創(chuàng)建消息隊列 起名：TestDirectQueue
* durable:是否持久化,默認是false,持久化隊列：會被存儲在磁盤上，當消息代理重啟時仍然存在，暫存隊列：當前連接有效
* exclusive:默認也是false，只能被當前創(chuàng)建的連接使用，而且當連接關閉后隊列即被刪除。此參考優(yōu)先級高于durable
* autoDelete:是否自動刪除，當沒有生產(chǎn)者或者消費者使用此隊列，該隊列會自動刪除。
* 一般設置一下隊列的持久化就好,其余兩個就是默認false
* @return new Queue(name,durable,exclusive,autoDelete)
*/
@Bean
public Queue TestDirectQueue() {
return new Queue("TestDirectQueue",true);
}
/**
* Direct交換機 起名：TestDirectExchange
*/
@Bean
DirectExchange TestDirectExchange() {
return new DirectExchange("TestDirectExchange",true,false);
}
/**
* 綁定
* 將隊列和交換機綁定, 并設置用于匹配鍵（路由鍵）：TestDirectRouting
*/
@Bean
Binding bindingDirect() {
return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting");
}
}
編寫controller層
@RestController
@RequestMapping("/rabbit")
public class SendMessageController {
@Autowired
private RabbitTemplate rabbitTemplate; //使用RabbitTemplate,這提供了接收/發(fā)送等等方法
@RequestMapping("/send")
public String sendDirectMessage() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "League of Legends never dies！";
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);
//將消息攜帶綁定鍵值：TestDirectRouting 發(fā)送到交換機TestDirectExchange
rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map);
return "ok";
}
}
把rabbitmq-provider項目運行，調(diào)用下接口:
因為我們目前還沒弄消費者 rabbitmq-consumer，消息沒有被消費的，我們?nèi)abbitMq管理頁面看看，是否推送成功：
消息已經(jīng)推送到rabbitMq服務器上面了。
2、創(chuàng)建rabbitmq-consumer項目：
pom.xml和application.yml跟上邊一樣（把端口號和應用名換了即可）
創(chuàng)建消息接收監(jiān)聽類，DirectReceiver01.java：
@Component
@RabbitListener(queues = "TestDirectQueue")//監(jiān)聽的隊列名稱 TestDirectQueue
public class DirectReceiver01 {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("DirectReceiver01消費者收到的消息: " + testMessage.toString());
}
}
啟動消費者項目：
運行結果如下
兩個消息直接被消費了。然后可以再繼續(xù)調(diào)用rabbitmq-provider項目的推送消息接口，可以看到消費者即時消費消息！
擴展
直連交換機既然是一對一，那如果咱們配置多臺監(jiān)聽綁定到同一個直連交互的同一個隊列，會怎么樣？
直接復制兩份DirectReceiver.java，改下打印內(nèi)容！
重新啟動項目！在生產(chǎn)者多發(fā)送幾次消息。運行結果如下：
可以看到是實現(xiàn)了輪詢的方式對消息進行消費，而且不存在重復消費。
2、Topic Exchange（主題交換機）
直連交換機的routing_key方案非常簡單，但是它是一對一的關系，那么我們需要一對多呢？希望一條消息發(fā)送給多個隊列。所以RabbitMQ提供了一種主題交換機，發(fā)送到主題交換機上的消息需要攜帶指定規(guī)則的routing_key，主題交換機會根據(jù)這個規(guī)則將數(shù)據(jù)發(fā)送到對應的(多個)隊列上。
主題交換機的routing_key需要有一定的規(guī)則，交換機和隊列的binding_key需要采用 *.# 的格式，每個部分用.分開，其中：
*表示一個單詞
#表示任意數(shù)量（零個或多個）單詞。（例如：topic.man）
1、在rabbitmq-provider項目里面創(chuàng)建TopicRabbitConfig.java：
@Configuration
public class TopicRabbitConfig {
//綁定鍵
public final static String man = "topic.man";
public final static String woman = "topic.woman";
@Bean
public Queue firstQueue() {
return new Queue(TopicRabbitConfig.man);
}
@Bean
public Queue secondQueue() {
return new Queue(TopicRabbitConfig.woman);
}
@Bean
TopicExchange exchange() {
return new TopicExchange("topicExchange");
}
//將firstQueue和topicExchange綁定,而且綁定的鍵值為topic.man
//這樣只要是消息攜帶的路由鍵是topic.man,才會分發(fā)到該隊列
@Bean
Binding bindingExchangeMessage() {
return BindingBuilder.bind(firstQueue()).to(exchange()).with(man);
}
//將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.#
// 這樣只要是消息攜帶的路由鍵是以topic.開頭,都會分發(fā)到該隊列
@Bean
Binding bindingExchangeMessage2() {
return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#");
}
}
編寫controller層
@RequestMapping("rabbit")
@RestController
public class TopicSendController {
@Autowired
private RabbitTemplate template;
@GetMapping("/sendTopicMessage1")
public String sendTopicMessage1() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: M A N ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> manMap = new HashMap<>();
manMap.put("messageId", messageId);
manMap.put("messageData", messageData);
manMap.put("createTime", createTime);
template.convertAndSend("topicExchange", "topic.man", manMap);
return "ok";
}
@GetMapping("/sendTopicMessage2")
public String sendTopicMessage2() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: woman is all ";
String createTime = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
Map<String, Object> womanMap = new HashMap<>();
womanMap.put("messageId", messageId);
womanMap.put("messageData", messageData);
womanMap.put("createTime", createTime);
template.convertAndSend("topicExchange", "topic.woman", womanMap);
return "ok";
}
}
不要急著去運行，先去編寫消費者,創(chuàng)建兩個接口
TopicManReceiver :
@Component
@RabbitListener(queues = "topic.man")
public class TopicManReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("TopicManReceiver消費者收到消息 : " + testMessage.toString());
}
}
TopicWoManReceiver :
@Component
@RabbitListener(queues = "topic.woman")
public class TopicWoManReceiver {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("TopicWoManReceiver消費者收到消息 : " + testMessage.toString());
}
}
然后把rabbitmq-provider，rabbitmq-consumer兩個項目都跑起來，先調(diào)用/sendTopicMessage1 接口：
然后看消費者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監(jiān)聽隊列1，綁定鍵為：topic.man
TopicTotalReceiver監(jiān)聽隊列2，綁定鍵為：topic.#
而當前推送的消息，攜帶的路由鍵為：topic.man
所以可以看到兩個監(jiān)聽消費者receiver都成功消費到了消息，因為這兩個recevier監(jiān)聽的隊列的綁定鍵都能與這條消息攜帶的路由鍵匹配上。
用/sendTopicMessage2接口：
然后看消費者rabbitmq-consumer的控制臺輸出情況：
TopicManReceiver監(jiān)聽隊列1，綁定鍵為：topic.man
TopicTotalReceiver監(jiān)聽隊列2，綁定鍵為：topic.#
而當前推送的消息，攜帶的路由鍵為：topic.woman
所以可以看到兩個監(jiān)聽消費者，只有TopicWoManReceiver 成功消費到了消息，
3、Fanout exchange（扇形交換機）
扇形交換機是最基本的交換機類型，它做的事情很簡單–廣播信息。Fanout交換機會把接收到的消息全部轉發(fā)到綁定的隊列上。因為廣播不需要“思考”，所以Fanout交換機是四種交換機中速度最快的。
同樣地，先在rabbitmq-provider項目上創(chuàng)建FanoutRabbitConfig.java：
@Configuration
public class FanoutRabbitConfig {
/**
* 創(chuàng)建三個隊列 ：fanout.A fanout.B fanout.C
* 將三個隊列都綁定在交換機 fanoutExchange 上
* 因為是扇型交換機, 路由鍵無需配置,配置也不起作用
*/
@Bean
public Queue queueA() {
return new Queue("fanout.A");
}
@Bean
public Queue queueB() {
return new Queue("fanout.B");
}
@Bean
public Queue queueC() {
return new Queue("fanout.C");
}
@Bean
FanoutExchange fanoutExchange() {
return new FanoutExchange("fanoutExchange");
}
@Bean
Binding bindingExchangeA() {
return BindingBuilder.bind(queueA()).to(fanoutExchange());
}
@Bean
Binding bindingExchangeB() {
return BindingBuilder.bind(queueB()).to(fanoutExchange());
}
@Bean
Binding bindingExchangeC() {
return BindingBuilder.bind(queueC()).to(fanoutExchange());
}
}
編寫controller層：
@RestController
@RequestMapping("rabbit")
public class FanoutController {
@Autowired
private RabbitTemplate template;
@GetMapping("/sendFanoutMessage")
public String sendFanoutMessage() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: testFanoutMessage ";
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);
template.convertAndSend("fanoutExchange", null, map);
return "ok";
}
}
接著在rabbitmq-consumer項目里加上消息消費類，
FanoutReceiverA:
@Component
@RabbitListener(queues = "fanout.A")
public class FanoutReceiverA {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("FanoutReceiverC消費者收到消息 : " +testMessage.toString());
}
}
FanoutReceiverB
@Component
@RabbitListener(queues = "fanout.B")
public class FanoutReceiverB{
@RabbitHandler
public void process(Map testMessage) {
System.out.println("FanoutReceiverC消費者收到消息 : " +testMessage.toString());
}
}
FanoutReceiverC :
@Component
@RabbitListener(queues = "fanout.C")
public class FanoutReceiverC {
@RabbitHandler
public void process(Map testMessage) {
System.out.println("FanoutReceiverC消費者收到消息 : " +testMessage.toString());
}
}
最后將rabbitmq-provider和rabbitmq-consumer項目都跑起來，調(diào)用下接口/sendFanoutMessage，運行結果如下：
可以看到只要發(fā)送到 fanoutExchange 這個扇型交換機的消息， 三個隊列都綁定這個交換機，所以三個消息接收類都監(jiān)聽到了這條消息。
4、消息確認（發(fā)送確認與接收確認） 4.1發(fā)送確認
在rabbitmq-provider項目中
編寫配置文件：
server:
port: 8090
spring:
application:
name: rabbitmq-consumer
rabbitmq:
host: 192.168.152.173
port: 5672
username: guest
password: guest
#確認消息已發(fā)送到交換機(Exchange)
publisher-confirm-type: correlated
#確認消息已發(fā)送到隊列(Queue)
publisher-returns: true
配置相關的消息確認回調(diào)函數(shù)，RabbitConfig.java：
/**
* 配置相關的消息確認回調(diào)函數(shù)
*/
@Configuration
public class RabbitConfig {
@Bean
public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) {
RabbitTemplate rabbitTemplate = new RabbitTemplate();
rabbitTemplate.setConnectionFactory(connectionFactory);
//設置開啟Mandatory,才能觸發(fā)回調(diào)函數(shù),無論消息推送結果怎么樣都強制調(diào)用回調(diào)函數(shù)
rabbitTemplate.setMandatory(true);
//確認回調(diào)
rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
@Override
public void confirm(CorrelationData correlationData, boolean condition, String cause) {
System.out.println("ConfirmCallback: " + "相關數(shù)據(jù)：" + correlationData);
System.out.println("ConfirmCallback: " + "確認情況：" + condition);
System.out.println("ConfirmCallback: " + "原因：" + cause);
}
});
//設置返回消息的回調(diào)
rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
@Override
public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
System.out.println("ReturnCallback: " + "消息：" + message);
System.out.println("ReturnCallback: " + "回應碼：" + replyCode);
System.out.println("ReturnCallback: " + "回應信息：" + replyText);
System.out.println("ReturnCallback: " + "交換機：" + exchange);
System.out.println("ReturnCallback: " + "路由鍵：" + routingKey);
}
});
return rabbitTemplate;
}
}
ConfirmCallback
ConfirmCallback是一個回調(diào)接口，就是只確認是否正確到達 Exchange 中。
ReturnCallback
通過實現(xiàn) ReturnCallback 接口，啟動消息失敗返回，此接口是在交換器路由不到隊列時觸發(fā)回調(diào)，該方法可以不使用，因為交換器和隊列是在代碼里綁定的，如果存在綁定隊列失敗，那就是你代碼寫錯了。
從總體的情況分析，推送消息存在四種情況：
1、消息推送到server，但是在server里找不到交換機
2、消息推送到server，找到交換機了，但是沒找到隊列
3、消息推送到sever，交換機和隊列啥都沒找到
4、消息推送成功
1、消息推送到server，但是在server里找不到交換機
寫個測試接口，把消息推送到名為‘non-existent-exchange'的交換機上（這個交換機是沒有創(chuàng)建沒有配置的）：
@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);
//non-existent-exchange這個交換機沒有創(chuàng)建和配置
template.convertAndSend("non-existent-exchange", "TestDirectRouting", map);
return "ok";
}
調(diào)用接口，查看rabbitmq-provuder項目的控制臺輸出情況
因為這個交換機沒有創(chuàng)建，消息不能到達到達 Exchange 中。所以會觸發(fā)ConfirmCallback這個回調(diào)函數(shù)
結論： 只會觸發(fā)ConfirmCallback。
2、消息推送到server，找到交換機了，但是沒找到隊列
新增一個交換機，但是不給這個交換機綁定隊列，在DirectRabitConfig里面新增一個直連交換機，不做任何綁定配置操作：
@Bean
DirectExchange lonelyDirectExchange() {
return new DirectExchange("lonelyDirectExchange");
}
同樣編寫測試接口
@GetMapping("/TestMessageAck2")
public String TestMessageAck2() {
String messageId = String.valueOf(UUID.randomUUID());
String messageData = "message: lonelyDirectExchange 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";
}
調(diào)用方法,查看運行結果：
可以看到兩個回調(diào)函數(shù)都被調(diào)用了；
這種情況下，消息是推送成功到服務器了的，所以ConfirmCallback對消息確認情況是true；
而在RetrunCallback回調(diào)函數(shù)的打印參數(shù)里面可以看到，消息是推送到了交換機成功了，但是在路由分發(fā)給隊列的時候，找不到隊列，所以報了錯誤 NO_ROUTE。
結論： 兩個回調(diào)函數(shù)都會觸發(fā)！
3、消息推送到sever，交換機和隊列啥都沒找到
結論： 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。
4、消息推送成功
結論： 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。
總結： ConfirmCallback回調(diào)函數(shù)總是會被觸發(fā)，確認情況true和false是有沒有找到交換機，false的話打印原因。ReturnCallback是交換機路由不到隊列是回調(diào)，也就是你交換機沒有綁定隊列，這個回調(diào)函數(shù)可以不寫，綁定失敗的話就好好看看代碼哪里寫錯了吧。
以上是生產(chǎn)者推送消息的消息確認 回調(diào)函數(shù)的使用介紹！
4.2 接受確認
消費者確認發(fā)生在監(jiān)聽隊列的消費者處理業(yè)務失敗，如，發(fā)生了異常，不符合要求的數(shù)據(jù)……，這些場景我們就需要手動處理，比如重新發(fā)送或者丟棄。
消息確認模式有：
- AcknowledgeMode.NONE：自動確認
- AcknowledgeMode.AUTO：根據(jù)情況確認
- AcknowledgeMode.MANUAL：手動確認
參考：https://blog.csdn.net/qq_35387940/article/details/100514134
4.1.1 自動確認
修改消費者配置文件
rabbitmq:
host: 192.168.152.173
port: 5672
username: guest
password: guest
listener:
direct:
acknowledge-mode: auto # manual：手動
在消費者項目里，
新建MessageListenerConfig.java上添加代碼相關的配置代碼：
package com.zsn.demo.config;
import com.zsn.demo.resceiver.MyAckReceiver;
import org.springframework.amqp.core.AcknowledgeMode;
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;
/**
* @author： zhouzhou
* @date：2021/7/22 9:49
*/
@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);
container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認是自動確認，這里改為手動確認消息
//設置一個隊列
container.setQueueNames("TestDirectQueue");
//如果同時設置多個如下： 前提是隊列都是必須已經(jīng)創(chuàng)建存在的
// container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3");
//另一種設置隊列的方法,如果使用這種情況,那么要設置多個,就使用addQueues
//container.setQueues(new Queue("TestDirectQueue",true));
//container.addQueues(new Queue("TestDirectQueue2",true));
//container.addQueues(new Queue("TestDirectQueue3",true));
container.setMessageListener(myAckReceiver);
return container;
}
}
對應的手動確認消息監(jiān)聽類，MyAckReceiver.java（手動確認模式需要實現(xiàn) ChannelAwareMessageListener）：
package com.zsn.demo.resceiver;
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;
/**
* @author： zhouzhou
* @date：2021/7/22 9:50
*/
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
/**
* 用于處理接收到的 Rabbit 消息的回調(diào)。
* 實現(xiàn)者應該處理指定的信息
* 通常通過給定的會話發(fā)送回復消息
* @param message the received AMQP message (never <code>null</code>)
* @param channel the underlying Rabbit Channel (never <code>null</code>)
* @throws Exception Any.
*/
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
String msg = message.toString();
String[] msgArray = msg.split("'");//可以點進Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
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 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime);
System.out.println("消費的主題消息來自："+message.getMessageProperties().getConsumerQueue());
channel.basicAck(deliveryTag, true); //第二個參數(shù)，手動確認可以被批處理，當該參數(shù)為 true 時，則可以一次性確認 delivery_tag 小于等于傳入值的所有消息
// channel.basicReject(deliveryTag, true);//第二個參數(shù)，true會重新放回隊列，所以需要自己根據(jù)業(yè)務邏輯判斷什么時候使用拒絕
} catch (Exception e) {
channel.basicReject(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 string : strs) {
String key = string.split("=")[0].trim();
String value = string.split("=")[1];
map.put(key, value);
}
return map;
}
}
需要注意的 basicAck 方法需要傳遞兩個參數(shù)
- deliveryTag（唯一標識 ID）：當一個消費者向 RabbitMQ 注冊后，會建立起一個 Channel ，RabbitMQ 會用 basic.deliver 方法向消費者推送消息，這個方法攜帶了一個 delivery tag， 它代表了 RabbitMQ 向該 Channel 投遞的這條消息的唯一標識 ID，是一個單調(diào)遞增的正整數(shù)，delivery tag 的范圍僅限于 Channel
- multiple：為了減少網(wǎng)絡流量，手動確認可以被批處理，當該參數(shù)為 true 時，則可以一次性確認 delivery_tag 小于等于傳入值的所有消息
basicNack方法需要傳遞三個參數(shù)
- deliveryTag（唯一標識 ID）：上面已經(jīng)解釋了。
- multiple：上面已經(jīng)解釋了。
- requeue： true ：重回隊列，false ：丟棄，我們在nack方法中必須設置 false，否則重發(fā)沒有意義。
basicReject方法需要傳遞兩個參數(shù)
- deliveryTag（唯一標識 ID）：上面已經(jīng)解釋了。
- requeue：上面已經(jīng)解釋了，在reject方法里必須設置true。
總結
- basic.ack用于肯定確認
- basic.nack用于否定確認（注意：這是AMQP 0-9-1的RabbitMQ擴展）
- basic.reject用于否定確認，但與basic.nack相比有一個限制:一次只能拒絕單條消息
消費者端以上的3個方法都表示消息已經(jīng)被正確投遞，但是basic.ack表示消息已經(jīng)被正確處理。
而basic.nack,basic.reject表示沒有被正確處理：
這時，先調(diào)用接口，給直連交換機TestDirectExchange 的隊列TestDirectQueue 推送一條消息，可以看到監(jiān)聽器正常消費了下來：
擴展1
除了直連交換機的隊列TestDirectQueue需要變成手動確認以外，我們還需要將一個其他的隊列或者多個隊列也變成手動確認，而且不同隊列實現(xiàn)不同的業(yè)務處理。
那么我們需要做的第一步，往SimpleMessageListenerContainer里添加多個隊列
MyAckReceiver.java 就可以同時將上面設置到的隊列的消息都消費下來。
@Component
public class MyAckReceiver implements ChannelAwareMessageListener {
/**
* 用于處理接收到的 Rabbit 消息的回調(diào)。
* 實現(xiàn)者應該處理指定的信息
* 通常通過給定的會話發(fā)送回復消息
* @param message the received AMQP message (never <code>null</code>)
* @param channel the underlying Rabbit Channel (never <code>null</code>)
* @throws Exception Any.
*/
@Override
public void onMessage(Message message, Channel channel) throws Exception {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
String msg = message.toString();
String[] msgArray = msg.split("'");//可以點進Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
Map<String, String> msgMap = mapStringToMap(msgArray[1].trim(),3);
String messageId=msgMap.get("messageId");
String messageData=msgMap.get("messageData");
String createTime=msgMap.get("createTime");
if ("TestDirectQueue".equals(message.getMessageProperties().getConsumerQueue())){
System.out.println("消費的消息來自的隊列名為："+message.getMessageProperties().getConsumerQueue());
System.out.println("消息成功消費到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime);
System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務處理流程......");
}
if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){
System.out.println("消費的消息來自的隊列名為："+message.getMessageProperties().getConsumerQueue());
System.out.println("消息成功消費到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime);
System.out.println("執(zhí)行fanout.A中的消息的業(yè)務處理流程......");
}
channel.basicAck(deliveryTag, true);
// channel.basicReject(deliveryTag, true);//為true會重新放回隊列
} catch (Exception e) {
channel.basicReject(deliveryTag, false);
e.printStackTrace();
}
}
//{key=value,key=value,key=value} 格式轉換成map
private Map<String, String> mapStringToMap(String str,int enNum) {
str = str.substring(1, str.length() - 1);
String[] strs = str.split(",",enNum);
Map<String, String> map = new HashMap<String, String>();
for (String string : strs) {
String key = string.split("=")[0].trim();
String value = string.split("=")[1];
map.put(key, value);
}
return map;
}
}
分別調(diào)用這兩個隊列相關接口，運行結果如下：
擴展2： 手動確認的另一種寫法：
我們在實際工作中可能不會這么寫，一般會寫在server層，自己定義一個接口：
RabbitConsumerService
public interface RabbitConsumerService {
void process(Channel channel, Message message) throws IOException;
}
@Component
public class RabbitConsumerServiceImpl implements RabbitConsumerService{
@RabbitHandler
@RabbitListener(queues = "TestDirectQueue")
@Override
public void process(Channel channel, Message message) throws IOException {
long deliveryTag = message.getMessageProperties().getDeliveryTag();
try {
//因為傳遞消息的時候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理
String msg = message.toString();
String[] msgArray = msg.split("'");//可以點進Message里面看源碼,單引號直接的數(shù)據(jù)就是我們的map消息數(shù)據(jù)
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 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime);
System.out.println("消費的主題消息來自："+message.getMessageProperties().getConsumerQueue());
channel.basicAck(deliveryTag, true); //第二個參數(shù)，手動確認可以被批處理，當該參數(shù)為 true 時，則可以一次性確認 delivery_tag 小于等于傳入值的所有消息
// channel.basicReject(deliveryTag, true);//第二個參數(shù)，true會重新放回隊列，所以需要自己根據(jù)業(yè)務邏輯判斷什么時候使用拒絕
} catch (Exception e) {
channel.basicReject(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 string : strs) {
String key = string.split("=")[0].trim();
String value = string.split("=")[1];
map.put(key, value);
}
return map;
}
}
@Component這個理論上要換成@service，但是由于我沒有添加日志，所以換成@service會報錯，這里就是提供一個思路，運行結果如下：
附：
rabbitmq在application.yml文件中的其他配置
rabbitmq:
addresses: 192.168.152.193:5672
username: guest
password: guest
#虛擬host 可以不設置,使用server默認host
virtual-host: /
listener:
simple:
prefetch: 1 #設置一次處理一個消息
acknowledge-mode: manual #設置消費端手動 ack
concurrency: 3 #設置同時有3個消費者消費
connection-timeout: 500
# 確認消息是否正確到達queue,如果沒有則觸發(fā)，如果有則不觸發(fā)
publisher-returns: true
#確認消息已發(fā)送到交換機(Exchange)
publisher-confirm-type: correlated
template:
#設置為 true后 消費者在消息沒有被路由到合適隊列情況下會被return監(jiān)聽，而不會自動刪除
mandatory: true
解釋：Mandatory 設置為 true 則會監(jiān)聽器會接受到路由不可達的消息，然后處理。如果設置為 false，broker 將會自動刪除該消息。
參考https://blog.csdn.net/qq_35387940/article/details/100514134
到此這篇關于RabbitMQ交換機與Springboot整合的簡單實現(xiàn)的文章就介紹到這了,更多相關RabbitMQ Springboot整合內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關文章希望大家以后多多支持腳本之家！
相關文章
spring boot實現(xiàn)profiles動態(tài)切換的示例
Spring Boot支持在不同的環(huán)境下使用不同的配置文件，該技術非常有利于持續(xù)集成，在構建項目的時候只需要使用不同的構建命令就可以生成不同運行環(huán)境下war包，而不需要手動切換配置文件。2020-10-10
最近在項目中使用了java中的split分割字符串，發(fā)現(xiàn)了一個bug，充分了展示了自己對java底層的認知有很多的不足和欠缺。下面將這次的經(jīng)過總結出來分享給大家，有需要的朋友們可以參考借鑒，下面來一起看看吧。2016-12-12
關于@ApiModel和@ApiModelProperty的使用
這篇文章主要介紹了關于@ApiModel和@ApiModelProperty的使用方式,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-11-11
這篇文章主要為大家介紹了apllo開源分布式配置中心部署詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪2023-02-02
這篇文章主要介紹了Java弱引用集合WeakHashMap總結,WeakHashMap利用WeakReference的弱引用特性讓用戶在使用的過程中不會因為沒有釋放Map中的資源而導致內(nèi)存泄露,WeakHashMap實現(xiàn)了Map接口,使用方式和其他的Map相同,需要的朋友可以參考下2023-09-09
這篇文章主要介紹了Spring Cloud Gateway去掉url前綴的操作，具有很好的參考價值，希望對大家有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2021-07-07
SpringCloud項目集成Feign、Hystrix過程解析
這篇文章主要介紹了SpringCloud項目集成Feign、Hystrix過程解析,文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值,需要的朋友可以參考下2019-11-11
mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在
這篇文章主要介紹了mybatis批量添加，批量更新之前如何判斷是否已經(jīng)存在，具有很好的參考價值，希望對的有所幫助。如有錯誤或未考慮完全的地方，望不吝賜教2022-08-08
這篇文章主要為大家詳細介紹了JAVA的觀察者模式，文中示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下，希望能夠給你帶來幫助2022-08-08
過濾器通常對一些web資源進行攔截，做完一些處理器再交給下一個過濾器處理，直到所有的過濾器處理器，再調(diào)用servlet實例的service方法進行處理。本文將通過示例為大家講解JavaWeb中過濾器Filter的用法與實現(xiàn)，需要的可以參考一下2022-08-08