说到 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. 架构

先来看一个简单的架构图：

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

1. 首先 Client 发送一条消息，和普通的消息相比，这条消息多了两个关键内容：一个是 correlation_id，这个表示这条消息的唯一 id，还有一个内容是 reply_to，这个表示消息回复队列的名字。
2. Server 从消息发送队列获取消息并处理相应的业务逻辑，处理完成后，将处理结果发送到 reply_to 指定的回调队列中。
3. Client 从回调队列中读取消息，就可以知道消息的执行情况是什么样子了。

这种情况其实非常适合处理异步调用。

2. 实践

创建一个新的Empty Project，工程名叫做mq_rpc

在这个mq_rpc工程中用Spring Initializr的方式创建两个SpringBoot的module

创建一个模块，模块名叫做server

选择依赖

然后以同样的方法在mq_rpc工程中创建一个新的模块，模块名叫做client，勾选的依赖同上

在mq_rpc工程的client模块的application.properties配置文件中作如下的配置

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 我们才能将发送的消息和返回值之间关联起来。最后一行配置则是开启发送失败退回。

在mq_rpc工程的client模块的org.javaboy.client.config包下面创建一个配置类

@Configuration
public class RabbitConfig {
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    public static final String RPC_REPLY_MSG_QUEUE = "rpc_reply_msg_queue";
    public static final String RPC_EXCHANGE = "rpc_exchange";
    @Bean
    Queue msgQueue() {
        return new Queue(RPC_MSG_QUEUE);
    }
    @Bean
    Queue replyQueue() {
        return new Queue(RPC_REPLY_MSG_QUEUE);
    }
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE);
    }
    @Bean
    Binding msgBinding() {
        return BindingBuilder.bind(msgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }
    @Bean
    Binding replyBinding() {
        return BindingBuilder.bind(replyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_MSG_QUEUE);
    }
    /**
     * 使用 RabbitTemplate发送和接收消息
     * 并设置回调队列地址
     */
    @Bean
    RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setReplyAddress(RPC_REPLY_MSG_QUEUE);
        rabbitTemplate.setReplyTimeout(6000);
        return rabbitTemplate;
    }
    /**
     * 给返回队列设置监听器
     */
    @Bean
    SimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {
        SimpleMessageListenerContainer simpleMessageListenerContainer = new SimpleMessageListenerContainer(connectionFactory);
        simpleMessageListenerContainer.setQueueNames(RPC_REPLY_MSG_QUEUE);
        simpleMessageListenerContainer.setMessageListener(rabbitTemplate(connectionFactory));
        return simpleMessageListenerContainer;
    }
}

这个配置类中我们分别配置了消息发送队列 msgQueue 和消息返回队列 replyQueue，然后将这两个队列和交换机进行绑定。这个都是 RabbitMQ 的常规操作，没啥好说的。

在 Spring Boot 中我们负责消息发送的工具是 RabbitTemplate，默认情况下，系统自动提供了该工具，但是这里我们需要对该工具重新进行定制，主要是添加消息发送的返回队列，最后我们还需要给返回队列设置一个监听器。

在mq_rpc工程的client模块的org.javaboy.client.controller包下创建一个控制器

@RestController
public class HelloController {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @GetMapping("/send")
    public void hello(String message) {
        //将来要发送的消息对象
        Message msg = MessageBuilder.withBody(message.getBytes()).build();
        //发送消息，方法的返回值就是 server 给出的响应
        Message result = rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_MSG_QUEUE, msg);
        if (result != null) {
            //发送的消息的 correlationId
            String correlationId = msg.getMessageProperties().getCorrelationId();
            //返回的消息的 correlationId
            String spring_returned_message_correlation = (String) result.getMessageProperties().getHeaders().get("spring_returned_message_correlation");
            if (correlationId.equals(spring_returned_message_correlation)) {
                System.out.println("收到服务端的响应：" + new String(result.getBody()));
            }
        }
    }
}

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

1. 消息发送调用 sendAndReceive 方法，该方法自带返回值，返回值就是服务端返回的消息。
2. 服务端返回的消息中，头信息中包含了 spring_returned_message_correlation 字段，这个就是消息发送时候的 correlation_id，通过消息发送时候的 correlation_id 以及返回消息头中的 spring_returned_message_correlation 字段值，我们就可以将返回的消息内容和发送的消息绑定到一起，确认出这个返回的内容就是针对这个发送的消息的。

这就是整个客户端的开发，其实最最核心的就是 sendAndReceive 方法的调用。调用虽然简单，但是准备工作还是要做足够。例如如果我们没有在 application.properties 中配置 correlated，发送的消息中就没有 correlation_id，这样就无法将返回的消息内容和发送的消息内容关联起来。

在mq_rpc工程的server模块的application.properties配置文件中作如下的配置

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
server.port=8081

在mq_rpc工程的server模块的org.javaboy.client.config包下面创建一个配置类，这个配置类就比较简单，单纯的配置一下消息队列并将之和消息交换机绑定起来，如下：

@Configuration
public class RabbitConfig {
    public static final String RPC_MSG_QUEUE = "rpc_msg_queue";
    public static final String RPC_REPLY_MSG_QUEUE = "rpc_reply_msg_queue";
    public static final String RPC_EXCHANGE = "rpc_exchange";
    @Bean
    Queue msgQueue() {
        return new Queue(RPC_MSG_QUEUE);
    }
    @Bean
    Queue replyQueue() {
        return new Queue(RPC_REPLY_MSG_QUEUE);
    }
    @Bean
    TopicExchange topicExchange() {
        return new TopicExchange(RPC_EXCHANGE);
    }
    @Bean
    Binding msgBinding() {
        return BindingBuilder.bind(msgQueue())
                .to(topicExchange())
                .with(RPC_MSG_QUEUE);
    }
    @Bean
    Binding replyBinding() {
        return BindingBuilder.bind(replyQueue())
                .to(topicExchange())
                .with(RPC_REPLY_MSG_QUEUE);
    }
}

在mq_rpc工程的server模块的org.javaboy.client.controller包下创建一个控制器

@Component
public class RpcServer {
    @Autowired
    RabbitTemplate rabbitTemplate;
    @RabbitListener(queues = RabbitConfig.RPC_MSG_QUEUE)
    public void process(Message message) {
        byte[] body = message.getBody();
        //这是服务端要返回的消息
        Message build = MessageBuilder.withBody(("我是服务端，我收到了客户端发来的：" + new String(body)).getBytes()).build();
        CorrelationData correlationData = new CorrelationData(message.getMessageProperties().getCorrelationId());
        rabbitTemplate.sendAndReceive(RabbitConfig.RPC_EXCHANGE, RabbitConfig.RPC_REPLY_MSG_QUEUE, build, correlationData);
    }
}

这里的逻辑就比较简单了：

1. 服务端首先收到消息并打印出来。
2. 服务端提取出原消息中的 correlation_id。
3. 服务端调用 sendAndReceive 方法，将消息发送给 rpc_reply_msg_queue队列，同时带上 correlation_id 参数。

服务端的消息发出后，客户端将收到服务端返回的结果。

启动两个服务，然后再浏览器访问：http://localhost:8080/send?message=江南一点雨，查看client应用的控制台