SpringBoot+RabbitMQ 實現 RPC 調用
說到 RPC(Remote Procedure Call Protocol 遠程過程調用協議)，小伙伴們腦海里蹦出的估計都是 RESTful API、Dubbo、WebService、Java RMI、CORBA 等。
其實，RabbitMQ 也給我們提供了 RPC 功能，并且使用起來很簡單。
今天松哥通過一個簡單的案例來和大家分享一下 Spring Boot+RabbitMQ 如何實現一個簡單的 RPC 調用。
注意
關于 RabbitMQ 實現 RPC 調用，有的小伙伴可能會有一些誤解，心想這還不簡單?搞兩個消息隊列 queue_1 和 queue_2，首先客戶端發送消息到 queue_1 上，服務端監聽 queue_1 上的消息，收到之后進行處理;處理完成后，服務端發送消息到 queue_2 隊列上，然后客戶端監聽 queue_2 隊列上的消息，這樣就知道服務端的處理結果了。
這種方式不是不可以，就是有點麻煩!RabbitMQ 中提供了現成的方案可以直接使用，非常方便。接下來我們就一起來學習下。
1. 架構
先來看一個簡單的架構圖：
這張圖把問題說的很明白了：
首先 Client 發送一條消息，和普通的消息相比，這條消息多了兩個關鍵內容：一個是 correlation_id，這個表示這條消息的唯一 id，還有一個內容是 reply_to，這個表示消息回復隊列的名字。
Server 從消息發送隊列獲取消息并處理相應的業務邏輯，處理完成后，將處理結果發送到 reply_to 指定的回調隊列中。
Client 從回調隊列中讀取消息，就可以知道消息的執行情況是什么樣子了。
這種情況其實非常適合處理異步調用。
2. 實踐
接下來我們通過一個具體的例子來看看這個怎么玩。
2.1 客戶端開發
首先我們來創建一個 Spring Boot 工程名為 producer，作為消息生產者，創建時候添加 web 和 rabbitmq 依賴，如下圖：
項目創建成功之后，首先在 application.properties 中配置 RabbitMQ 的基本信息，如下：
- spring.rabbitmq.host=localhost
- spring.rabbitmq.port=5672
- spring.rabbitmq.username=guest
- spring.rabbitmq.password=guest
- spring.rabbitmq.publisher-confirm-type=correlated
- spring.rabbitmq.publisher-returns=true
這個配置前面四行都好理解，我就不贅述，后面兩行：首先是配置消息確認方式，我們通過 correlated 來確認，只有開啟了這個配置，將來的消息中才會帶 correlation_id，只有通過 correlation_id 我們才能將發送的消息和返回值之間關聯起來。最后一行配置則是開啟發送失敗退回。
接下來我們來提供一個配置類，如下：
- /**
- * @author 江南一點雨
- * @微信公眾號 江南一點雨
- * @網站 http://www.itboyhub.com
- * @國際站 http://www.javaboy.org
- * @微信 a_java_boy
- * @GitHub https://github.com/lenve
- * @Gitee https://gitee.com/lenve
- */
- @Configuration
- public class RabbitConfig {
- public static final String RPC_QUEUE1 = "queue_1";
- public static final String RPC_QUEUE2 = "queue_2";
- public static final String RPC_EXCHANGE = "rpc_exchange";
- /**
- * 設置消息發送RPC隊列
- */
- @Bean
- Queue msgQueue() {
- return new Queue(RPC_QUEUE1);
- }
- /**
- * 設置返回隊列
- */
- @Bean
- Queue replyQueue() {
- return new Queue(RPC_QUEUE2);
- }
- /**
- * 設置交換機
- */
- @Bean
- TopicExchange exchange() {
- return new TopicExchange(RPC_EXCHANGE);
- }
- /**
- * 請求隊列和交換器綁定
- */
- @Bean
- Binding msgBinding() {
- return BindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);
- }
- /**
- * 返回隊列和交換器綁定
- */
- @Bean
- Binding replyBinding() {
- return BindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);
- }
- /**
- * 使用 RabbitTemplate發送和接收消息
- * 并設置回調隊列地址
- */
- @Bean
- RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
- RabbitTemplate template = new RabbitTemplate(connectionFactory);
- template.setReplyAddress(RPC_QUEUE2);
- template.setReplyTimeout(6000);
- return template;
- }
- /**
- * 給返回隊列設置監聽器
- */
- @Bean
- SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {
- SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
- container.setConnectionFactory(connectionFactory);
- container.setQueueNames(RPC_QUEUE2);
- container.setMessageListener(rabbitTemplate(connectionFactory));
- return container;
- }
- }
這個配置類中我們分別配置了消息發送隊列 msgQueue 和消息返回隊列 replyQueue，然后將這兩個隊列和消息交換機進行綁定。這個都是 RabbitMQ 的常規操作，沒啥好說的。
在 Spring Boot 中我們負責消息發送的工具是 RabbitTemplate，默認情況下，系統自動提供了該工具，但是這里我們需要對該工具重新進行定制，主要是添加消息發送的返回隊列，最后我們還需要給返回隊列設置一個監聽器。
好啦，接下來我們就可以開始具體的消息發送了：
- /**
- * @author 江南一點雨
- * @微信公眾號 江南一點雨
- * @網站 http://www.itboyhub.com
- * @國際站 http://www.javaboy.org
- * @微信 a_java_boy
- * @GitHub https://github.com/lenve
- * @Gitee https://gitee.com/lenve
- */
- @RestController
- public class RpcClientController {
- private static final Logger logger = LoggerFactory.getLogger(RpcClientController.class);
- @Autowired
- private RabbitTemplate rabbitTemplate;
- @GetMapping("/send")
- public String send(String message) {
- // 創建消息對象
- Message newMessage = MessageBuilder.withBody(message.getBytes()).build();
- logger.info("client send：{}", newMessage);
- //客戶端發送消息
- Message result = rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_QUEUE1, newMessage);
- String response = "";
- if (result != null) {
- // 獲取已發送的消息的 correlationId
- String correlationId = newMessage.getMessageProperties().getCorrelationId();
- logger.info("correlationId:{}", correlationId);
- // 獲取響應頭信息
- HashMap<String, Object> headers = (HashMap<String, Object>) result.getMessageProperties().getHeaders();
- // 獲取 server 返回的消息 id
- String msgId = (String) headers.get("spring_returned_message_correlation");
- if (msgId.equals(correlationId)) {
- response = new String(result.getBody());
- logger.info("client receive：{}", response);
- }
- }
- return response;
- }
- }
這塊的代碼其實也都是一些常規代碼，我挑幾個關鍵的節點說下：
- 消息發送調用 sendAndReceive 方法，該方法自帶返回值，返回值就是服務端返回的消息。
- 服務端返回的消息中，頭信息中包含了 spring_returned_message_correlation 字段，這個就是消息發送時候的 correlation_id，通過消息發送時候的 correlation_id 以及返回消息頭中的 spring_returned_message_correlation 字段值，我們就可以將返回的消息內容和發送的消息綁定到一起，確認出這個返回的內容就是針對這個發送的消息的。
這就是整個客戶端的開發，其實最最核心的就是 sendAndReceive 方法的調用。調用雖然簡單，但是準備工作還是要做足夠。例如如果我們沒有在 application.properties 中配置 correlated，發送的消息中就沒有 correlation_id，這樣就無法將返回的消息內容和發送的消息內容關聯起來。
2.2 服務端開發
再來看看服務端的開發。
首先創建一個名為 consumer 的 Spring Boot 項目，創建項目添加的依賴和客戶端開發創建的依賴是一致的，不再贅述。
然后配置 application.properties 配置文件，該文件的配置也和客戶端中的配置一致，不再贅述。
接下來提供一個 RabbitMQ 的配置類，這個配置類就比較簡單，單純的配置一下消息隊列并將之和消息交換機綁定起來，如下：
- /**
- * @author 江南一點雨
- * @微信公眾號 江南一點雨
- * @網站 http://www.itboyhub.com
- * @國際站 http://www.javaboy.org
- * @微信 a_java_boy
- * @GitHub https://github.com/lenve
- * @Gitee https://gitee.com/lenve
- */
- @Configuration
- public class RabbitConfig {
- public static final String RPC_QUEUE1 = "queue_1";
- public static final String RPC_QUEUE2 = "queue_2";
- public static final String RPC_EXCHANGE = "rpc_exchange";
- /**
- * 配置消息發送隊列
- */
- @Bean
- Queue msgQueue() {
- return new Queue(RPC_QUEUE1);
- }
- /**
- * 設置返回隊列
- */
- @Bean
- Queue replyQueue() {
- return new Queue(RPC_QUEUE2);
- }
- /**
- * 設置交換機
- */
- @Bean
- TopicExchange exchange() {
- return new TopicExchange(RPC_EXCHANGE);
- }
- /**
- * 請求隊列和交換器綁定
- */
- @Bean
- Binding msgBinding() {
- return BindingBuilder.bind(msgQueue()).to(exchange()).with(RPC_QUEUE1);
- }
- /**
- * 返回隊列和交換器綁定
- */
- @Bean
- Binding replyBinding() {
- return BindingBuilder.bind(replyQueue()).to(exchange()).with(RPC_QUEUE2);
- }
- }
最后我們再來看下消息的消費：
- @Component
- public class RpcServerController {
- private static final Logger logger = LoggerFactory.getLogger(RpcServerController.class);
- @Autowired
- private RabbitTemplate rabbitTemplate;
- @RabbitListener(queues = RabbitConfig.RPC_QUEUE1)
- public void process(Message msg) {
- logger.info("server receive : {}",msg.toString());
- Message response = MessageBuilder.withBody(("i'm receive:"+new String(msg.getBody())).getBytes()).build();
- CorrelationData correlationData = new CorrelationData(msg.getMessageProperties().getCorrelationId());
- rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_QUEUE2, response, correlationData);
- }
- }
這里的邏輯就比較簡單了：
- 服務端首先收到消息并打印出來。
- 服務端提取出原消息中的 correlation_id。
- 服務端調用 sendAndReceive 方法，將消息發送給 RPC_QUEUE2 隊列，同時帶上 correlation_id 參數。
服務端的消息發出后，客戶端將收到服務端返回的結果。
OK，大功告成。
2.3 測試
接下來我們進行一個簡單測試。
首先啟動 RabbitMQ。
接下來分別啟動 producer 和 consumer，然后在 postman 中調用 producer 的接口進行測試，如下：
可以看到，已經收到了服務端的返回信息。
來看看 producer 的運行日志：
可以看到，消息發送出去后，同時也收到了 consumer 返回的信息。
可以看到，consumer 也收到了客戶端發來的消息。
3. 小結
好啦，一個小小的案例，帶小伙伴們體驗一把 RabbitMQ 實現 RPC 調用。
