国产精品久久久久一区二区三区,日韩在线观看免费完整版视频,国产精品国产三级国产专不∫

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. 架構

先來看一個簡單的架構圖：

SpringBoot+RabbitMQ 實現 RPC 調用

這張圖把問題說的很明白了：

首先 Client 發送一條消息，和普通的消息相比，這條消息多了兩個關鍵內容：一個是 correlation_id，這個表示這條消息的唯一 id，還有一個內容是 reply_to，這個表示消息回復隊列的名字。

Server 從消息發送隊列獲取消息并處理相應的業務邏輯，處理完成后，將處理結果發送到 reply_to 指定的回調隊列中。

Client 從回調隊列中讀取消息，就可以知道消息的執行情況是什么樣子了。

這種情況其實非常適合處理異步調用。

2. 實踐

接下來我們通過一個具體的例子來看看這個怎么玩。

2.1 客戶端開發

首先我們來創建一個 Spring Boot 工程名為 producer，作為消息生產者，創建時候添加 web 和 rabbitmq 依賴，如下圖：

SpringBoot+RabbitMQ 實現 RPC 調用

項目創建成功之后，首先在 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 headers = (HashMap) 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);
}
}