MQ的作用包括了解耦、非同步等。
通常生產者只負責生產訊息,而不關心訊息誰去獲取,或消費結果如何;消費者只負責接收指定的訊息進行業務處理而不關心訊息從哪裡來一級回覆業務處理情況。但我們專案中有特殊的業務存在,我們作為消息生產者在生產消息後需要接收消費者的響應結果(說白了就是類似同步調用請求響應的MQ使用),經過研究,MQ的Reply模式(直接回复模式)就是為此種業務模式而產生。
依賴:
##我這裡只列出最核心的rabbitMq所需依賴<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-amqp</artifactId> </dependency>
配置:
無其餘特殊配置,因為reply就是rabbitmq的一種互動方式而已spring: rabbitmq: host: 10.50.40.116 port: 5673 username: admin password: admin(2 )RabbitMq bean設定
package com.leilei.demo; import org.springframework.amqp.core.FanoutExchange; import org.springframework.amqp.core.Queue; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; /** * @author lei * @create 2022-09-19 21:44 * @desc mq配置 **/ @Configuration public class RabbitMqConfig { @Bean public Queue bizQueue() { return new Queue("bizQueue"); } @Bean public Queue replyQueue() { return new Queue("replyQueue"); } @Bean FanoutExchange bizExchange() { return new FanoutExchange("bizExchange"); } }
業務類別:
@Data @NoArgsConstructor @AllArgsConstructor public class Vehicle implements Serializable { private Integer id; private String name; }(3)訊息生產端訊息生產端需要做的事情:有生產訊息、接受訊息消費回應
(1)生產訊息
/** * 生产消息 * * @param * @return void * @author lei * @date 2022-09-19 21:59:18 */ public void replySend() { MessageProperties messageProperties = new MessageProperties(); messageProperties.setReplyTo("replyQueue"); //todo 根据业务,做一个严谨的全局唯一ID,我这里暂时用UUID String correlationId = UUID.randomUUID().toString(); // 我这里指定了唯一消息ID,看业务场景,消费者消费响应后,生产者端可根据消息ID做业务处理 messageProperties.setCorrelationId(correlationId); Vehicle vehicle = new Vehicle(1, "川A0001"); Message message = new Message(JSON.toJSONString(vehicle).getBytes(), messageProperties); rabbitTemplate.convertAndSend("bizExchange","",message); System.out.println("生产者发送消息,自定义消息ID为:" + correlationId); }
(2)接受Reply回應
消費者消費訊息後會將處理結果傳送到一個佇列,我們讀取這裡佇列就可以拿到對應訊息的回應結果進行業務處理了/** * 接收消息响应 * * @param message * @return void * @author lei * @date 2022-09-19 21:59:27 */ @RabbitListener(queues = "replyQueue") public void replyResponse(Message message) { String s = new String(message.getBody()); String correlationId = message.getMessageProperties().getCorrelationId(); System.out.println("收到客户端响应消息ID:" + correlationId); //todo 根据消息ID可判断这是哪一个消息的响应,我们就可做业务操作 System.out.println("收到客户端响应消息:" + s); }(4)訊息消費端訊息消費端需要做的事有:接受訊息然後進行業務處理、回應訊息(1)方法一:sendTo註解方法傳回值一般來說,我們mq消費者監聽方法不需要回傳值,我們這裡使用sendTo註解,則需要將要回應的訊息定義為回傳值,sendTo註解中指定要回應到哪個佇列
重點:
/** * 方式1 SendTo指定响应队列 * * @param message * @return String * @author lei * @date 2022-09-19 16:17:52 */ @RabbitListener(queues ="bizQueue") @SendTo("replyQueue") public String handleEmailMessage(Message message) { try { String msg=new String(message.getBody(), StandardCharsets.UTF_8); log.info("---consumer接收到消息----{}",msg); return "客户端响应消息:"+msg+"处理完成!"; } catch (Exception e) { log.error("处理业务消息失败",e); } return null; }(2)方法二:讀取生產端的訊息使用範本發送與普通的消費者方法一樣,只需要RabbitListener註解監聽業務佇列;但還需要根據訊息取得ReplyTo位址,然後自己消費者方法內部手動傳送訊息
/** * 方式2 message消息获取内部reply rabbitmq手动发送 * * @param message * @return String * @author lei * @date 2022-09-19 16:17:52 */ @RabbitListener(queues = "bizQueue") public void handleEmailMessage2(Message message) { try { String msg = new String(message.getBody(), StandardCharsets.UTF_8); log.info("---consumer接收到消息----{}", msg); String replyTo = message.getMessageProperties().getReplyTo(); System.out.println("接收到的reply:" + replyTo); rabbitTemplate.convertAndSend(replyTo, "客户端响应消息:" + msg + "处理完成!", x -> { x.getMessageProperties().setCorrelationId(message.getMessageProperties().getCorrelationId()); return x; }); } catch (Exception e) { log.error("处理业务消息失败",e); } }
/** * 方式三 方法有返回值,返回要响应的数据 (reply 由生产者发送消息时指定,消费者不做任何处理) * * @param message * @return String * @author lei * @date 2022-09-19 23:17:47 */ @RabbitListener(queues ="bizQueue") public String handleEmailMessage3(Message message) { try { String msg=new String(message.getBody(), StandardCharsets.UTF_8); log.info("---consumer接收到消息----{}",msg); return "客户端响应消息:"+msg+"处理完成!"; } catch (Exception e) { log.error("处理业务消息失败",e); } return null; }
(4)測試
生產訊息:消費訊息與回應:
############## ###############收到的回應:#####################連結:####### ##########
以上是springboot rabbitmq reply訊息直接回覆模式怎麼實現的詳細內容。更多資訊請關注PHP中文網其他相關文章!