前言
我们之前介绍了RabbitMQ的五种模型（详见上方系列文章《RabbitMQ灵活运用，怎么理解五种消息模型》），即简单、轮询、主题、发布/订阅、路由、主题五种模式。除此之外，rabbitMQ还提供了一种 RPC 模式，这种模式是怎么回事？真的好用吗？一起来了解下
一、RPC 及 RabbitMQ-RPC模型
1. RPC概念
RPC是指远程过程调用（Remote Procedure Call），是一种计算机通信协议，用于将一个计算机程序的执行过程转移至另一台计算机上，但对用户而言，它就像是在本地运行一样，RPC通常用于分布式系统或组件化架构中，可以隐藏分布式系统背后的复杂性
2. RabbitMQ-RPC模型
我们照例先看官方流程图
RPC 工作模式如下：
- 当客户端启动时，它会创建一个匿名独占回调队列。
- 对于 RPC 请求，客户端发送具有两个属性的消息：reply_to，设置为回调队列，correlation_id，设置为每个请求的唯一值，请求将发送到rpc_queue队列。
- RPC 工作线程（又名：服务器）正在等待该队列上的请求。当请求出现时，它会执行作业，并使用 reply_to 字段中的队列将包含结果的消息发送回客户端。
- 客户端等待回调队列中的数据。出现消息时，它会检查 correlation_id 属性。如果它与请求中的值匹配，则它将响应返回到应用程序。
可以看出，所谓RPC模式其实是两台机器互为消息生产者和消费者，同时利用两条队列，一个负责传递参数，一个负责传递结果
二、RPC模式的Demo
在进一步探讨前，我们先实际运行一段demo
1. 调用方代码
在这个示例中，我们创建了一个RPCClient类，并在构造函数中建立了RabbitMQ连接并创建了一个通道。我们还声明了一个请求队列名称requestQueueName和一个回复队列名称replyQueueName。
在call方法中，我们将请求消息发布到请求队列，并使用replyTo属性将回复队列名称包括在消息中。接着我们定义一个response阻塞队列，它将用于存储来自回复队列的响应。最后，我们使用BlockingQueue.take()方法等待响应内容并返回结果。
调用方代码如下（示例）：
package com.example.seeu.mq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.TimeoutException; public class RPCClient implements AutoCloseable { private final Connection connection; private final Channel channel; private final String requestQueueName = "rpc_queue"; private String replyQueueName; final BlockingQueue<String> response = new ArrayBlockingQueue<>(1); public RPCClient() throws IOException, TimeoutException { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); connection = factory.newConnection(); channel = connection.createChannel(); replyQueueName = channel.queueDeclare().getQueue(); channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> { response.offer(new String(delivery.getBody(), "UTF-8")); }, consumerTag -> { }); } public String call(String message) throws IOException, InterruptedException { final String corrId = UUID.randomUUID().toString(); AMQP.BasicProperties props = new AMQP.BasicProperties .Builder() .correlationId(corrId) .replyTo(replyQueueName) .build(); channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8")); String result = response.take(); return result; } public void close() throws IOException { connection.close(); } public static void main(String[] argv) { try (RPCClient fibonacciRpc = new RPCClient()) { for (int i = 1; i <= 30; i++) { System.out.println(" [x] 请求 fib("+ i +")"); String response = fibonacciRpc.call(String.valueOf(i)); System.out.println(" [.] fib("+ i +")的结果为 '" + response + "'"); } } catch (IOException | TimeoutException | InterruptedException e) { e.printStackTrace(); } } }
2. 服务方代码
在被调用方，声明了一个名为RPC_QUEUE_NAME的请求队列，并使用queueDeclare方法在RabbitMQ中声明它。我们还设置了basicQos，以确保每次只接收一个请求。
在main方法中，我们定义了一个DeliverCallback，它将处理每个传入的消息并生成响应。对于每个传入的消息，我们提取出correlationId和请求消息。我们执行一些计算，然后将结果作为字符串回复到名为replyTo的消息队列中。最后，我们使用basicAck确认我们已收到消息。
我们使用basicConsume在请求队列上注册一个消费者，并将DeliverCallback传递给它。我们还使用synchronized块和wait()方法来等待每个传入的消息处理完成并立即发送回复。
被调用方代码：
package com.example.seeu.mq; import com.rabbitmq.client.*; import java.io.IOException; import java.util.concurrent.TimeoutException; public class RPCServer { private static final String RPC_QUEUE_NAME = "rpc_queue"; private static int fib(int n) { if (n == 0) return 0; if (n == 1) return 1; return fib(n - 1) + fib(n - 2); } public static void main(String[] argv) { ConnectionFactory factory = new ConnectionFactory(); factory.setHost("localhost"); try (Connection connection = factory.newConnection(); Channel channel = connection.createChannel()) { channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null); channel.basicQos(1); System.out.println(" [x] Awaiting RPC requests"); Object monitor = new Object(); // 当成功接受到消息时，进行的处理逻辑 DeliverCallback deliverCallback = (consumerTag, delivery) -> { AMQP.BasicProperties replyProps = new AMQP.BasicProperties .Builder() // 返回参数中需要附带请求的id .correlationId(delivery.getProperties().getCorrelationId()) .build(); String response = ""; try { String message = new String(delivery.getBody(), "UTF-8"); int n = Integer.parseInt(message); System.out.println(" [.] fib(" + message + ")"); response += fib(n); } catch (RuntimeException e) { System.out.println(" [.] " + e.toString()); } finally { // delivery.getProperties().getReplyTo() 即为返回的队列，将结果发送给该该队列 channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8")); // 接受确认，通知RabbitMQ 服务器自己已经消费掉此条消息 channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false); synchronized (monitor) { monitor.notify(); } } }; // 推模式，监听获取消息 channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { })); // 主线程活跃 while (true) { synchronized (monitor) { try { monitor.wait(); } catch (InterruptedException e) { e.printStackTrace(); } } } } catch (IOException | TimeoutException e) { e.printStackTrace(); } } }
结果：
以上示例仅为最简单的 1 对 1 同步调用的示例，当调用方和被调用方是多对多的关系时，调用方还得考虑获取的结果乱序的问题，必须为每一个请求找到其对应的结果
三、RabbitMQ-RPC 的利与弊
尽管我们现在知道了 RabbitMQ 有RPC 的用法，但是其跟真正的 RPC 框架(如duubo) 等进行对比，又有什么优劣呢？
优势：
Dubbo 的跨语言支持相对较差，主要支持 Java 、Golang及部分其他语言；RabbitMQ 支持的语言更多
劣势：
RabbitMQ周边支持不完善，如序列化问题，同步异步选择，服务治理，都需要开发者自行设计与编码
调用方和被调用方多对多时，结果集可能乱序，需开发者为每个结果寻找其请求线程并返回值
高度依赖MQ集群的可用性和稳定性，一旦MQ故障，RPC即无法进行
总结：和真正的RPC框架比起来，优势不足，缺陷太多，主要是需要开发者进行大量代码编写才能实现较完善的RPC功能，有这时间，不如直接使用现成的RPC框架了，所以虽然RabbitMQ 官方提供了 RPC 模式，但使用者寥寥。除非是少量的异步调用，或跨语言问题无法解决，否则笔者亦不建议在生产上使用RabbitMQ 作为 RPC 框架
