说到 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. 实际

接下来咱们通过一个具体的例子来看看这个怎么玩。

2.1 客户端开发

首先咱们来创立一个 Spring Boot 工程名为 producer，作为音讯生产者，创立时候增加 web 和 rabbitmq 依赖，如下图：

我的项目创立胜利之后，首先在 application.properties 中配置 RabbitMQ 的根本信息，如下：

spring.rabbitmq.host=localhostspring.rabbitmq.port=5672spring.rabbitmq.username=guestspring.rabbitmq.password=guestspring.rabbitmq.publisher-confirm-type=correlatedspring.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 */@Configurationpublic 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 */@RestControllerpublic 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;    }}

这块的代码其实也都是一些惯例代码，我挑几个要害的节点说下：

  1. 音讯发送调用 sendAndReceive 办法，该办法自带返回值，返回值就是服务端返回的音讯。
  2. 服务端返回的音讯中，头信息中蕴含了 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 */@Configurationpublic 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);    }}

最初咱们再来看下音讯的生产：

@Componentpublic 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);    }}

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

  1. 服务端首先收到音讯并打印进去。
  2. 服务端提取出原音讯中的 correlation_id。
  3. 服务端调用 sendAndReceive 办法，将音讯发送给 RPC_QUEUE2 队列，同时带上 correlation_id 参数。

服务端的音讯收回后，客户端将收到服务端返回的后果。

OK，功败垂成。

2.3 测试

接下来咱们进行一个简略测试。

首先启动 RabbitMQ。

接下来别离启动 producer 和 consumer，而后在 postman 中调用 producer 的接口进行测试，如下：

能够看到，曾经收到了服务端的返回信息。

来看看 producer 的运行日志：

能够看到，音讯发送进来后，同时也收到了 consumer 返回的信息。

能够看到，consumer 也收到了客户端发来的音讯。

3. 小结

好啦，一个小小的案例，带小伙伴们体验一把 RabbitMQ 实现 RPC 调用。

公众号江南一点雨后盾回复 mq_rpc 能够获取本文案例哦～感兴趣的小伙伴能够试试～