RabbitMQ交換機(jī)與Springboot整合的簡(jiǎn)單實(shí)現(xiàn)
RabbitMQ-交換機(jī)
1、交換機(jī)是干什么的?
消息(Message)由Client發(fā)送,RabbitMQ接收到消息之后通過(guò)交換機(jī)轉(zhuǎn)發(fā)到對(duì)應(yīng)的隊(duì)列上面。Worker會(huì)從隊(duì)列中獲取未被讀取的數(shù)據(jù)處理。
1、交換機(jī)的種類(lèi)
RabbitMQ包含四種不同的交換機(jī)類(lèi)型:
- Direct exchange:直連交換機(jī),轉(zhuǎn)發(fā)消息到routigKey指定的隊(duì)列
- Fanout exchange:扇形交換機(jī),轉(zhuǎn)發(fā)消息到所有綁定隊(duì)列(速度最快)
- Topic exchange:主題交換機(jī),按規(guī)則轉(zhuǎn)發(fā)消息(最靈活)
- Headers exchange:首部交換機(jī) (不常用)
1、Direct exchange(直連交換機(jī))
直連交換機(jī)是一種帶路由功能的交換機(jī),根據(jù)消息攜帶的路由鍵將消息投遞給對(duì)應(yīng)隊(duì)列。
一個(gè)隊(duì)列通過(guò)routing_key(路由鍵)與一個(gè)交換機(jī)綁定,當(dāng)消息被發(fā)送的時(shí)候,需要指定一個(gè)routing_key,這個(gè)消息被送達(dá)交換機(jī)的時(shí)候,就會(huì)被交換機(jī)送到指定的隊(duì)列里面去。
類(lèi)似一對(duì)一的關(guān)系!
1.1 Springboot的簡(jiǎn)單實(shí)現(xiàn)
創(chuàng)建2個(gè)springboot項(xiàng)目,一個(gè) rabbitmq-provider (生產(chǎn)者),一個(gè)rabbitmq-consumer(消費(fèi)者)。
1、首先創(chuàng)建 rabbitmq-provider
pom.xml添加依賴(lài):
<!--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頁(yè)面端口,消息發(fā)送端口為5672 username: guest password: guest
編寫(xiě)配置文件
@Configuration public class DirectRabbitConfig { /** * 創(chuàng)建消息隊(duì)列 起名:TestDirectQueue * durable:是否持久化,默認(rèn)是false,持久化隊(duì)列:會(huì)被存儲(chǔ)在磁盤(pán)上,當(dāng)消息代理重啟時(shí)仍然存在,暫存隊(duì)列:當(dāng)前連接有效 * exclusive:默認(rèn)也是false,只能被當(dāng)前創(chuàng)建的連接使用,而且當(dāng)連接關(guān)閉后隊(duì)列即被刪除。此參考優(yōu)先級(jí)高于durable * autoDelete:是否自動(dòng)刪除,當(dāng)沒(méi)有生產(chǎn)者或者消費(fèi)者使用此隊(duì)列,該隊(duì)列會(huì)自動(dòng)刪除。 * 一般設(shè)置一下隊(duì)列的持久化就好,其余兩個(gè)就是默認(rèn)false * @return new Queue(name,durable,exclusive,autoDelete) */ @Bean public Queue TestDirectQueue() { return new Queue("TestDirectQueue",true); } /** * Direct交換機(jī) 起名:TestDirectExchange */ @Bean DirectExchange TestDirectExchange() { return new DirectExchange("TestDirectExchange",true,false); } /** * 綁定 * 將隊(duì)列和交換機(jī)綁定, 并設(shè)置用于匹配鍵(路由鍵):TestDirectRouting */ @Bean Binding bindingDirect() { return BindingBuilder.bind(TestDirectQueue()).to(TestDirectExchange()).with("TestDirectRouting"); } }
編寫(xiě)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ā)送到交換機(jī)TestDirectExchange rabbitTemplate.convertAndSend("TestDirectExchange", "TestDirectRouting", map); return "ok"; } }
把rabbitmq-provider項(xiàng)目運(yùn)行,調(diào)用下接口:
因?yàn)槲覀兡壳斑€沒(méi)弄消費(fèi)者 rabbitmq-consumer,消息沒(méi)有被消費(fèi)的,我們?nèi)abbitMq管理頁(yè)面看看,是否推送成功:
消息已經(jīng)推送到rabbitMq服務(wù)器上面了。
2、創(chuàng)建rabbitmq-consumer項(xiàng)目:
pom.xml和application.yml跟上邊一樣(把端口號(hào)和應(yīng)用名換了即可)
創(chuàng)建消息接收監(jiān)聽(tīng)類(lèi),DirectReceiver01.java:
@Component @RabbitListener(queues = "TestDirectQueue")//監(jiān)聽(tīng)的隊(duì)列名稱(chēng) TestDirectQueue public class DirectReceiver01 { @RabbitHandler public void process(Map testMessage) { System.out.println("DirectReceiver01消費(fèi)者收到的消息: " + testMessage.toString()); } }
啟動(dòng)消費(fèi)者項(xiàng)目:
運(yùn)行結(jié)果如下
兩個(gè)消息直接被消費(fèi)了。然后可以再繼續(xù)調(diào)用rabbitmq-provider項(xiàng)目的推送消息接口,可以看到消費(fèi)者即時(shí)消費(fèi)消息!
擴(kuò)展
直連交換機(jī)既然是一對(duì)一,那如果咱們配置多臺(tái)監(jiān)聽(tīng)綁定到同一個(gè)直連交互的同一個(gè)隊(duì)列,會(huì)怎么樣?
直接復(fù)制兩份DirectReceiver.java,改下打印內(nèi)容!
重新啟動(dòng)項(xiàng)目!在生產(chǎn)者多發(fā)送幾次消息。運(yùn)行結(jié)果如下:
可以看到是實(shí)現(xiàn)了輪詢(xún)的方式對(duì)消息進(jìn)行消費(fèi),而且不存在重復(fù)消費(fèi)。
2、Topic Exchange(主題交換機(jī))
直連交換機(jī)的routing_key方案非常簡(jiǎn)單,但是它是一對(duì)一的關(guān)系,那么我們需要一對(duì)多呢?希望一條消息發(fā)送給多個(gè)隊(duì)列。所以RabbitMQ提供了一種主題交換機(jī),發(fā)送到主題交換機(jī)上的消息需要攜帶指定規(guī)則的routing_key,主題交換機(jī)會(huì)根據(jù)這個(gè)規(guī)則將數(shù)據(jù)發(fā)送到對(duì)應(yīng)的(多個(gè))隊(duì)列上。
主題交換機(jī)的routing_key需要有一定的規(guī)則,交換機(jī)和隊(duì)列的binding_key需要采用 *.# 的格式,每個(gè)部分用.分開(kāi),其中:
*表示一個(gè)單詞
#表示任意數(shù)量(零個(gè)或多個(gè))單詞。(例如:topic.man)
1、在rabbitmq-provider項(xiàng)目里面創(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,才會(huì)分發(fā)到該隊(duì)列 @Bean Binding bindingExchangeMessage() { return BindingBuilder.bind(firstQueue()).to(exchange()).with(man); } //將secondQueue和topicExchange綁定,而且綁定的鍵值為用上通配路由鍵規(guī)則topic.# // 這樣只要是消息攜帶的路由鍵是以topic.開(kāi)頭,都會(huì)分發(fā)到該隊(duì)列 @Bean Binding bindingExchangeMessage2() { return BindingBuilder.bind(secondQueue()).to(exchange()).with("topic.#"); } }
編寫(xiě)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"; } }
不要急著去運(yùn)行,先去編寫(xiě)消費(fèi)者,創(chuàng)建兩個(gè)接口
TopicManReceiver :
@Component @RabbitListener(queues = "topic.man") public class TopicManReceiver { @RabbitHandler public void process(Map testMessage) { System.out.println("TopicManReceiver消費(fèi)者收到消息 : " + testMessage.toString()); } }
TopicWoManReceiver :
@Component @RabbitListener(queues = "topic.woman") public class TopicWoManReceiver { @RabbitHandler public void process(Map testMessage) { System.out.println("TopicWoManReceiver消費(fèi)者收到消息 : " + testMessage.toString()); } }
然后把rabbitmq-provider,rabbitmq-consumer兩個(gè)項(xiàng)目都跑起來(lái),先調(diào)用/sendTopicMessage1 接口:
然后看消費(fèi)者rabbitmq-consumer的控制臺(tái)輸出情況:
TopicManReceiver監(jiān)聽(tīng)隊(duì)列1,綁定鍵為:topic.man
TopicTotalReceiver監(jiān)聽(tīng)隊(duì)列2,綁定鍵為:topic.#
而當(dāng)前推送的消息,攜帶的路由鍵為:topic.man
所以可以看到兩個(gè)監(jiān)聽(tīng)消費(fèi)者receiver都成功消費(fèi)到了消息,因?yàn)檫@兩個(gè)recevier監(jiān)聽(tīng)的隊(duì)列的綁定鍵都能與這條消息攜帶的路由鍵匹配上。
用/sendTopicMessage2接口:
然后看消費(fèi)者rabbitmq-consumer的控制臺(tái)輸出情況:
TopicManReceiver監(jiān)聽(tīng)隊(duì)列1,綁定鍵為:topic.man
TopicTotalReceiver監(jiān)聽(tīng)隊(duì)列2,綁定鍵為:topic.#
而當(dāng)前推送的消息,攜帶的路由鍵為:topic.woman
所以可以看到兩個(gè)監(jiān)聽(tīng)消費(fèi)者,只有TopicWoManReceiver 成功消費(fèi)到了消息,
3、Fanout exchange(扇形交換機(jī))
扇形交換機(jī)是最基本的交換機(jī)類(lèi)型,它做的事情很簡(jiǎn)單–廣播信息。Fanout交換機(jī)會(huì)把接收到的消息全部轉(zhuǎn)發(fā)到綁定的隊(duì)列上。因?yàn)閺V播不需要“思考”,所以Fanout交換機(jī)是四種交換機(jī)中速度最快的。
同樣地,先在rabbitmq-provider項(xiàng)目上創(chuàng)建FanoutRabbitConfig.java:
@Configuration public class FanoutRabbitConfig { /** * 創(chuàng)建三個(gè)隊(duì)列 :fanout.A fanout.B fanout.C * 將三個(gè)隊(duì)列都綁定在交換機(jī) fanoutExchange 上 * 因?yàn)槭巧刃徒粨Q機(jī), 路由鍵無(wú)需配置,配置也不起作用 */ @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()); } }
編寫(xiě)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項(xiàng)目里加上消息消費(fèi)類(lèi),
FanoutReceiverA:
@Component @RabbitListener(queues = "fanout.A") public class FanoutReceiverA { @RabbitHandler public void process(Map testMessage) { System.out.println("FanoutReceiverC消費(fèi)者收到消息 : " +testMessage.toString()); } }
FanoutReceiverB
@Component @RabbitListener(queues = "fanout.B") public class FanoutReceiverB{ @RabbitHandler public void process(Map testMessage) { System.out.println("FanoutReceiverC消費(fèi)者收到消息 : " +testMessage.toString()); } }
FanoutReceiverC :
@Component @RabbitListener(queues = "fanout.C") public class FanoutReceiverC { @RabbitHandler public void process(Map testMessage) { System.out.println("FanoutReceiverC消費(fèi)者收到消息 : " +testMessage.toString()); } }
最后將rabbitmq-provider和rabbitmq-consumer項(xiàng)目都跑起來(lái),調(diào)用下接口/sendFanoutMessage,運(yùn)行結(jié)果如下:
可以看到只要發(fā)送到 fanoutExchange 這個(gè)扇型交換機(jī)的消息, 三個(gè)隊(duì)列都綁定這個(gè)交換機(jī),所以三個(gè)消息接收類(lèi)都監(jiān)聽(tīng)到了這條消息。
4、消息確認(rèn)(發(fā)送確認(rèn)與接收確認(rèn)) 4.1發(fā)送確認(rèn)
在rabbitmq-provider項(xiàng)目中
編寫(xiě)配置文件:
server: port: 8090 spring: application: name: rabbitmq-consumer rabbitmq: host: 192.168.152.173 port: 5672 username: guest password: guest #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange) publisher-confirm-type: correlated #確認(rèn)消息已發(fā)送到隊(duì)列(Queue) publisher-returns: true
配置相關(guān)的消息確認(rèn)回調(diào)函數(shù),RabbitConfig.java:
/** * 配置相關(guān)的消息確認(rèn)回調(diào)函數(shù) */ @Configuration public class RabbitConfig { @Bean public RabbitTemplate createRabbitTemplate(ConnectionFactory connectionFactory) { RabbitTemplate rabbitTemplate = new RabbitTemplate(); rabbitTemplate.setConnectionFactory(connectionFactory); //設(shè)置開(kāi)啟Mandatory,才能觸發(fā)回調(diào)函數(shù),無(wú)論消息推送結(jié)果怎么樣都強(qiáng)制調(diào)用回調(diào)函數(shù) rabbitTemplate.setMandatory(true); //確認(rèn)回調(diào) rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() { @Override public void confirm(CorrelationData correlationData, boolean condition, String cause) { System.out.println("ConfirmCallback: " + "相關(guān)數(shù)據(jù):" + correlationData); System.out.println("ConfirmCallback: " + "確認(rèn)情況:" + condition); System.out.println("ConfirmCallback: " + "原因:" + cause); } }); //設(shè)置返回消息的回調(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: " + "回應(yīng)碼:" + replyCode); System.out.println("ReturnCallback: " + "回應(yīng)信息:" + replyText); System.out.println("ReturnCallback: " + "交換機(jī):" + exchange); System.out.println("ReturnCallback: " + "路由鍵:" + routingKey); } }); return rabbitTemplate; } }
ConfirmCallback
ConfirmCallback是一個(gè)回調(diào)接口,就是只確認(rèn)是否正確到達(dá) Exchange 中。
ReturnCallback
通過(guò)實(shí)現(xiàn) ReturnCallback 接口,啟動(dòng)消息失敗返回,此接口是在交換器路由不到隊(duì)列時(shí)觸發(fā)回調(diào),該方法可以不使用,因?yàn)榻粨Q器和隊(duì)列是在代碼里綁定的,如果存在綁定隊(duì)列失敗,那就是你代碼寫(xiě)錯(cuò)了。
從總體的情況分析,推送消息存在四種情況:
1、消息推送到server,但是在server里找不到交換機(jī)
2、消息推送到server,找到交換機(jī)了,但是沒(méi)找到隊(duì)列
3、消息推送到sever,交換機(jī)和隊(duì)列啥都沒(méi)找到
4、消息推送成功
1、消息推送到server,但是在server里找不到交換機(jī)
寫(xiě)個(gè)測(cè)試接口,把消息推送到名為‘non-existent-exchange'的交換機(jī)上(這個(gè)交換機(jī)是沒(méi)有創(chuàng)建沒(méi)有配置的):
@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這個(gè)交換機(jī)沒(méi)有創(chuàng)建和配置 template.convertAndSend("non-existent-exchange", "TestDirectRouting", map); return "ok"; }
調(diào)用接口,查看rabbitmq-provuder項(xiàng)目的控制臺(tái)輸出情況
因?yàn)檫@個(gè)交換機(jī)沒(méi)有創(chuàng)建,消息不能到達(dá)到達(dá) Exchange 中。所以會(huì)觸發(fā)ConfirmCallback這個(gè)回調(diào)函數(shù)
結(jié)論: 只會(huì)觸發(fā)ConfirmCallback。
2、消息推送到server,找到交換機(jī)了,但是沒(méi)找到隊(duì)列
新增一個(gè)交換機(jī),但是不給這個(gè)交換機(jī)綁定隊(duì)列,在DirectRabitConfig里面新增一個(gè)直連交換機(jī),不做任何綁定配置操作:
@Bean DirectExchange lonelyDirectExchange() { return new DirectExchange("lonelyDirectExchange"); }
同樣編寫(xiě)測(cè)試接口
@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)用方法,查看運(yùn)行結(jié)果:
可以看到兩個(gè)回調(diào)函數(shù)都被調(diào)用了;
這種情況下,消息是推送成功到服務(wù)器了的,所以ConfirmCallback對(duì)消息確認(rèn)情況是true;
而在RetrunCallback回調(diào)函數(shù)的打印參數(shù)里面可以看到,消息是推送到了交換機(jī)成功了,但是在路由分發(fā)給隊(duì)列的時(shí)候,找不到隊(duì)列,所以報(bào)了錯(cuò)誤 NO_ROUTE。
結(jié)論: 兩個(gè)回調(diào)函數(shù)都會(huì)觸發(fā)!
3、消息推送到sever,交換機(jī)和隊(duì)列啥都沒(méi)找到
結(jié)論: 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。
4、消息推送成功
結(jié)論: 觸發(fā)的是 ConfirmCallback 回調(diào)函數(shù)。
總結(jié): ConfirmCallback回調(diào)函數(shù)總是會(huì)被觸發(fā),確認(rèn)情況true和false是有沒(méi)有找到交換機(jī),false的話(huà)打印原因。ReturnCallback是交換機(jī)路由不到隊(duì)列是回調(diào),也就是你交換機(jī)沒(méi)有綁定隊(duì)列,這個(gè)回調(diào)函數(shù)可以不寫(xiě),綁定失敗的話(huà)就好好看看代碼哪里寫(xiě)錯(cuò)了吧。
以上是生產(chǎn)者推送消息的消息確認(rèn) 回調(diào)函數(shù)的使用介紹!
4.2 接受確認(rèn)
消費(fèi)者確認(rèn)發(fā)生在監(jiān)聽(tīng)隊(duì)列的消費(fèi)者處理業(yè)務(wù)失敗,如,發(fā)生了異常,不符合要求的數(shù)據(jù)……,這些場(chǎng)景我們就需要手動(dòng)處理,比如重新發(fā)送或者丟棄。
消息確認(rèn)模式有:
- AcknowledgeMode.NONE:自動(dòng)確認(rèn)
- AcknowledgeMode.AUTO:根據(jù)情況確認(rèn)
- AcknowledgeMode.MANUAL:手動(dòng)確認(rèn)
參考:https://blog.csdn.net/qq_35387940/article/details/100514134
4.1.1 自動(dòng)確認(rèn)
修改消費(fèi)者配置文件
rabbitmq: host: 192.168.152.173 port: 5672 username: guest password: guest listener: direct: acknowledge-mode: auto # manual:手動(dòng)
在消費(fèi)者項(xiàng)目里,
新建MessageListenerConfig.java上添加代碼相關(guān)的配置代碼:
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;//消息接收處理類(lèi) @Bean public SimpleMessageListenerContainer simpleMessageListenerContainer() { SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory); container.setConcurrentConsumers(1); container.setMaxConcurrentConsumers(1); container.setAcknowledgeMode(AcknowledgeMode.MANUAL); // RabbitMQ默認(rèn)是自動(dòng)確認(rèn),這里改為手動(dòng)確認(rèn)消息 //設(shè)置一個(gè)隊(duì)列 container.setQueueNames("TestDirectQueue"); //如果同時(shí)設(shè)置多個(gè)如下: 前提是隊(duì)列都是必須已經(jīng)創(chuàng)建存在的 // container.setQueueNames("TestDirectQueue","TestDirectQueue2","TestDirectQueue3"); //另一種設(shè)置隊(duì)列的方法,如果使用這種情況,那么要設(shè)置多個(gè),就使用addQueues //container.setQueues(new Queue("TestDirectQueue",true)); //container.addQueues(new Queue("TestDirectQueue2",true)); //container.addQueues(new Queue("TestDirectQueue3",true)); container.setMessageListener(myAckReceiver); return container; } }
對(duì)應(yīng)的手動(dòng)確認(rèn)消息監(jiān)聽(tīng)類(lèi),MyAckReceiver.java(手動(dòng)確認(rèn)模式需要實(shí)現(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)。 * 實(shí)現(xiàn)者應(yīng)該處理指定的信息 * 通常通過(guò)給定的會(huì)話(huà)發(fā)送回復(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 { //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理 String msg = message.toString(); String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號(hào)直接的數(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("消費(fèi)的主題消息來(lái)自:"+message.getMessageProperties().getConsumerQueue()); channel.basicAck(deliveryTag, true); //第二個(gè)參數(shù),手動(dòng)確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時(shí),則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息 // channel.basicReject(deliveryTag, true);//第二個(gè)參數(shù),true會(huì)重新放回隊(duì)列,所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時(shí)候使用拒絕 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } //{key=value,key=value,key=value} 格式轉(zhuǎn)換成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 方法需要傳遞兩個(gè)參數(shù)
- deliveryTag(唯一標(biāo)識(shí) ID):當(dāng)一個(gè)消費(fèi)者向 RabbitMQ 注冊(cè)后,會(huì)建立起一個(gè) Channel ,RabbitMQ 會(huì)用 basic.deliver 方法向消費(fèi)者推送消息,這個(gè)方法攜帶了一個(gè) delivery tag, 它代表了 RabbitMQ 向該 Channel 投遞的這條消息的唯一標(biāo)識(shí) ID,是一個(gè)單調(diào)遞增的正整數(shù),delivery tag 的范圍僅限于 Channel
- multiple:為了減少網(wǎng)絡(luò)流量,手動(dòng)確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時(shí),則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息
basicNack方法需要傳遞三個(gè)參數(shù)
- deliveryTag(唯一標(biāo)識(shí) ID):上面已經(jīng)解釋了。
- multiple:上面已經(jīng)解釋了。
- requeue: true :重回隊(duì)列,false :丟棄,我們?cè)趎ack方法中必須設(shè)置 false,否則重發(fā)沒(méi)有意義。
basicReject方法需要傳遞兩個(gè)參數(shù)
- deliveryTag(唯一標(biāo)識(shí) ID):上面已經(jīng)解釋了。
- requeue:上面已經(jīng)解釋了,在reject方法里必須設(shè)置true。
總結(jié)
- basic.ack用于肯定確認(rèn)
- basic.nack用于否定確認(rèn)(注意:這是AMQP 0-9-1的RabbitMQ擴(kuò)展)
- basic.reject用于否定確認(rèn),但與basic.nack相比有一個(gè)限制:一次只能拒絕單條消息
消費(fèi)者端以上的3個(gè)方法都表示消息已經(jīng)被正確投遞,但是basic.ack表示消息已經(jīng)被正確處理。
而basic.nack,basic.reject表示沒(méi)有被正確處理:
這時(shí),先調(diào)用接口,給直連交換機(jī)TestDirectExchange 的隊(duì)列TestDirectQueue 推送一條消息,可以看到監(jiān)聽(tīng)器正常消費(fèi)了下來(lái):
擴(kuò)展1
除了直連交換機(jī)的隊(duì)列TestDirectQueue需要變成手動(dòng)確認(rèn)以外,我們還需要將一個(gè)其他的隊(duì)列或者多個(gè)隊(duì)列也變成手動(dòng)確認(rèn),而且不同隊(duì)列實(shí)現(xiàn)不同的業(yè)務(wù)處理。
那么我們需要做的第一步,往SimpleMessageListenerContainer里添加多個(gè)隊(duì)列
MyAckReceiver.java 就可以同時(shí)將上面設(shè)置到的隊(duì)列的消息都消費(fèi)下來(lái)。
@Component public class MyAckReceiver implements ChannelAwareMessageListener { /** * 用于處理接收到的 Rabbit 消息的回調(diào)。 * 實(shí)現(xiàn)者應(yīng)該處理指定的信息 * 通常通過(guò)給定的會(huì)話(huà)發(fā)送回復(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 { //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理 String msg = message.toString(); String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號(hào)直接的數(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("消費(fèi)的消息來(lái)自的隊(duì)列名為:"+message.getMessageProperties().getConsumerQueue()); System.out.println("消息成功消費(fèi)到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); System.out.println("執(zhí)行TestDirectQueue中的消息的業(yè)務(wù)處理流程......"); } if ("fanout.A".equals(message.getMessageProperties().getConsumerQueue())){ System.out.println("消費(fèi)的消息來(lái)自的隊(duì)列名為:"+message.getMessageProperties().getConsumerQueue()); System.out.println("消息成功消費(fèi)到 messageId:"+messageId+" messageData:"+messageData+" createTime:"+createTime); System.out.println("執(zhí)行fanout.A中的消息的業(yè)務(wù)處理流程......"); } channel.basicAck(deliveryTag, true); // channel.basicReject(deliveryTag, true);//為true會(huì)重新放回隊(duì)列 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } //{key=value,key=value,key=value} 格式轉(zhuǎn)換成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)用這兩個(gè)隊(duì)列相關(guān)接口,運(yùn)行結(jié)果如下:
擴(kuò)展2: 手動(dòng)確認(rèn)的另一種寫(xiě)法:
我們?cè)趯?shí)際工作中可能不會(huì)這么寫(xiě),一般會(huì)寫(xiě)在server層,自己定義一個(gè)接口:
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 { //因?yàn)閭鬟f消息的時(shí)候用的map傳遞,所以將Map從Message內(nèi)取出需要做些處理 String msg = message.toString(); String[] msgArray = msg.split("'");//可以點(diǎn)進(jìn)Message里面看源碼,單引號(hào)直接的數(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("消費(fèi)的主題消息來(lái)自:"+message.getMessageProperties().getConsumerQueue()); channel.basicAck(deliveryTag, true); //第二個(gè)參數(shù),手動(dòng)確認(rèn)可以被批處理,當(dāng)該參數(shù)為 true 時(shí),則可以一次性確認(rèn) delivery_tag 小于等于傳入值的所有消息 // channel.basicReject(deliveryTag, true);//第二個(gè)參數(shù),true會(huì)重新放回隊(duì)列,所以需要自己根據(jù)業(yè)務(wù)邏輯判斷什么時(shí)候使用拒絕 } catch (Exception e) { channel.basicReject(deliveryTag, false); e.printStackTrace(); } } //{key=value,key=value,key=value} 格式轉(zhuǎn)換成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這個(gè)理論上要換成@service,但是由于我沒(méi)有添加日志,所以換成@service會(huì)報(bào)錯(cuò),這里就是提供一個(gè)思路,運(yùn)行結(jié)果如下:
附:
rabbitmq在application.yml文件中的其他配置
rabbitmq: addresses: 192.168.152.193:5672 username: guest password: guest #虛擬host 可以不設(shè)置,使用server默認(rèn)host virtual-host: / listener: simple: prefetch: 1 #設(shè)置一次處理一個(gè)消息 acknowledge-mode: manual #設(shè)置消費(fèi)端手動(dòng) ack concurrency: 3 #設(shè)置同時(shí)有3個(gè)消費(fèi)者消費(fèi) connection-timeout: 500 # 確認(rèn)消息是否正確到達(dá)queue,如果沒(méi)有則觸發(fā),如果有則不觸發(fā) publisher-returns: true #確認(rèn)消息已發(fā)送到交換機(jī)(Exchange) publisher-confirm-type: correlated template: #設(shè)置為 true后 消費(fèi)者在消息沒(méi)有被路由到合適隊(duì)列情況下會(huì)被return監(jiān)聽(tīng),而不會(huì)自動(dòng)刪除 mandatory: true
解釋?zhuān)篗andatory 設(shè)置為 true 則會(huì)監(jiān)聽(tīng)器會(huì)接受到路由不可達(dá)的消息,然后處理。如果設(shè)置為 false,broker 將會(huì)自動(dòng)刪除該消息。
參考https://blog.csdn.net/qq_35387940/article/details/100514134
到此這篇關(guān)于RabbitMQ交換機(jī)與Springboot整合的簡(jiǎn)單實(shí)現(xiàn)的文章就介紹到這了,更多相關(guān)RabbitMQ Springboot整合內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家!
相關(guān)文章
spring boot實(shí)現(xiàn)profiles動(dòng)態(tài)切換的示例
Spring Boot支持在不同的環(huán)境下使用不同的配置文件,該技術(shù)非常有利于持續(xù)集成,在構(gòu)建項(xiàng)目的時(shí)候只需要使用不同的構(gòu)建命令就可以生成不同運(yùn)行環(huán)境下war包,而不需要手動(dòng)切換配置文件。2020-10-10java用split分割字符串的一個(gè)有趣現(xiàn)象
最近在項(xiàng)目中使用了java中的split分割字符串,發(fā)現(xiàn)了一個(gè)bug,充分了展示了自己對(duì)java底層的認(rèn)知有很多的不足和欠缺。下面將這次的經(jīng)過(guò)總結(jié)出來(lái)分享給大家,有需要的朋友們可以參考借鑒,下面來(lái)一起看看吧。2016-12-12關(guān)于@ApiModel和@ApiModelProperty的使用
這篇文章主要介紹了關(guān)于@ApiModel和@ApiModelProperty的使用方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2023-11-11SpringCloud項(xiàng)目集成Feign、Hystrix過(guò)程解析
這篇文章主要介紹了SpringCloud項(xiàng)目集成Feign、Hystrix過(guò)程解析,文中通過(guò)示例代碼介紹的非常詳細(xì),對(duì)大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友可以參考下2019-11-11mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在
這篇文章主要介紹了mybatis批量添加,批量更新之前如何判斷是否已經(jīng)存在,具有很好的參考價(jià)值,希望對(duì)的有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-08-08Java設(shè)計(jì)模式探究之觀(guān)察者模式詳解
這篇文章主要為大家詳細(xì)介紹了JAVA的觀(guān)察者模式,文中示例代碼介紹的非常詳細(xì),具有一定的參考價(jià)值,感興趣的小伙伴們可以參考一下,希望能夠給你帶來(lái)幫助2022-08-08