出门莫恨无人随，书中车马多如簇。这篇文章主要讲述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. 架构先来看一个简单的架构图：

文章图片

这张图把问题说的很明白了：

2. 首先 Client 发送一条消息，和普通的消息相比，这条消息多了两个关键内容：一个是 correlation_id，这个表示这条消息的唯一 id，还有一个内容是 reply_to，这个表示消息回复队列的名字。
3. Server 从消息发送队列获取消息并处理相应的业务逻辑，处理完成后，将处理结果发送到 reply_to 指定的回调队列中。
4. 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;
}
}

【SpringBoot+RabbitMQ 实现 RPC 调用】这个配置类中我们分别配置了消息发送队列 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;
}
}

这块的代码其实也都是一些常规代码，我挑几个关键的节点说下：

2. 消息发送调用 sendAndReceive 方法，该方法自带返回值，返回值就是服务端返回的消息。
3. 服务端返回的消息中，头信息中包含了 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(("im receive:"

推荐阅读

• 

  大衬衣怎么搭配 让人眼前一亮
• 

  玻璃胶如何使用？玻璃胶如何使用?
• 

  橙子是什么季节的水果 怎么挑选橙子
• 

  华为手机忘记锁屏密码怎么办
• 

  如何快速升级公益服务器？ 公益服务器怎么升级最快
• 

  苹果动图如何保存 苹果怎么保存贴吧动图
• 

  志高空调无法开机该怎么解决?,基本上用这几种方法
• 

  国庆节购车指南 2020国庆节买车优惠力度大吗
• 

  立式空调怎么消毒
• 

  华为S7703交换机,一个镜像口如何对应多个观察口？华为S7703的介绍
• 

  灵芝孢子粉怎么吃
• 

  linux的echo命令 linux中echo命令的用法
• 

  李白的《清平调》是李白最有意境的诗吗？
• 

  最新版excel怎么制作迷你图 excel2016怎么添加迷你图
• 

  湖北荆州一周天气预报_湖北荆州天气查询
• 

  新款混动汉兰达报价及图片 新款汉兰达混动版落地多少钱
• 

  驱动桥类型 驱动桥类型有哪几种
• 

  文明6全面开战神级难度教学 全面开战怎么玩
• 

  玩咖女是什么意思
• 

  android中shape 的使用

• 存储结构与管理硬盘（part3）
• 存储结构与管理硬盘（part4）
• 过了35岁还有必要学习linux吗
• gst 测试摄像头命令
• Nginx使用X-Frame-Options防止被iframe跨域
• centos7离线安装htop
• JMeter性能测试中如何使用“用户参数”实现参数化
• LINUX学习第十九天
• mac下将python2.7改为python3