RabbitMQ RPC(Remote procedure call) 云计算集群的远程调用

本系列教程主要来自于官网入门教程的翻译，然后自己进行了部分的修改与实验，内容仅供参考。

官网地址  https://www.rabbitmq.com/tutorials/tutorial-six-java.html

在第二个教程中我们学习了如何使用工作队列分配耗时的任务在多个工人。
但是,如果我们需要在远程计算机上运行一个函数并等待结果?嗯,这是一个不同的故事。这种模式是俗称Remote
Procedure Call或RPC。

在本教程中我们将使用RabbitMQ来构建一个RPC系统:一个客户端和一个可伸缩的RPC服务器。我们没有任何耗时任务价值分配,我们要创建一个虚拟的RPC服务,返回斐波纳契数列。

1. 客户端接口 Client interface

        为了展示一个RPC服务是如何使用的，我们将创建一段很简单的客户端class。 它将会向外提供名字为call的函数，这个call会发送RPC请求并且阻塞知道收到RPC运算的结果。代码如下：

FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);

2. 回调函数队列 Callback queue

        总体来说，在RabbitMQ进行RPC远程调用是比较容易的。client发送请求的Message然后server返回响应结果。为了收到响应client在publish message时需要提供一个”callback“（回调）的queue地址。code如下:

callbackQueueName = channel.queueDeclare().getQueue();

BasicProperties props = new BasicProperties
.Builder()
.replyTo(callbackQueueName)
.build();

channel.basicPublish("", "rpc_queue", props, message.getBytes());

2.1 Message properties

AMQP 预定义了14个属性。它们中的绝大多很少会用到。以下几个是平时用的比较多的：

delivery_mode: 持久化一个Message（通过设定值为2）。其他任意值都是非持久化。请移步RabbitMQ消息队列（三）：任务分发机制
content_type: 描述mime-type 的encoding。比如设置为JSON编码：设置该property为application/json。
reply_to: 一般用来指明用于回调的queue（Commonly used to name a callback queue）。
correlation_id: 在请求中关联处理RPC响应（correlate RPC responses with requests）。

3. 相关id Correlation id

       在上个小节里，实现方法是对每个RPC请求都会创建一个callback queue。这是不高效的。幸运的是，在这里有一个解决方法：为每个client创建唯一的callback queue。

       这又有其他问题了：收到响应后它无法确定是否是它的，因为所有的响应都写到同一个queue了。上一小节的correlation_id在这种情况下就派上用场了：对于每个request，都设置唯一的一个值，在收到响应后，通过这个值就可以判断是否是自己的响应。如果不是自己的响应，就不去处理。

4. 总结

                    


 工作流程：
当客户端启动时，它创建了匿名的exclusive callback queue.
客户端的RPC请求时将同时设置两个properties： reply_to设置为callback queue；correlation_id设置为每个request一个独一无二的值.
请求将被发送到an rpc_queue queue.
RPC端或者说server一直在等待那个queue的请求。当请求到达时，它将通过在reply_to指定的queue回复一个message给client。
client一直等待callback queue的数据。当message到达时，它将检查correlation_id的值，如果值和它request发送时的一致那么就将返回响应。

5. 最终实现

The code for RPCServer.java

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;

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) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

connection = factory.newConnection();
channel = connection.createChannel();

channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

channel.basicQos(1);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

System.out.println(" [x] Awaiting RPC requests");

while (true) {
String response = null;

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

BasicProperties props = delivery.getProperties();
BasicProperties replyProps = new BasicProperties
.Builder()
.correlationId(props.getCorrelationId())
.build();

try {
String message = new String(delivery.getBody(),"UTF-8");
int n = Integer.parseInt(message);

System.out.println(" [.] fib(" + message + ")");
response = "" + fib(n);
}
catch (Exception e){
System.out.println(" [.] " + e.toString());
response = "";
}
finally {
channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes("UTF-8"));

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
catch  (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
}

服务器代码相当简单:
(1)像往常一样,我们首先建立连接和声明队列。
(2)我们fibonacci函数声明。它假定只有有效输入正整数。(别指望这一个庞大的数字,它可能是最慢递归实现可能的)。
(3)我们为basic_consume声明回调,RPC服务器的核心。在收到请求时执行。它的工作并发送响应。
(4)我们可能希望运行多个服务器进程。为了传播负载同样在多个服务器,我们需要设置theprefetch_count设置。

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.AMQP.BasicProperties;

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) {
Connection connection = null;
Channel channel = null;
try {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

connection = factory.newConnection();
channel = connection.createChannel();

channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

channel.basicQos(1);

QueueingConsumer consumer = new QueueingConsumer(channel);
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);

System.out.println(" [x] Awaiting RPC requests");

while (true) {
String response = null;

QueueingConsumer.Delivery delivery = consumer.nextDelivery();

BasicProperties props = delivery.getProperties();
BasicProperties replyProps = new BasicProperties
.Builder()
.correlationId(props.getCorrelationId())
.build();

try {
String message = new String(delivery.getBody(),"UTF-8");
int n = Integer.parseInt(message);

System.out.println(" [.] fib(" + message + ")");
response = "" + fib(n);
}
catch (Exception e){
System.out.println(" [.] " + e.toString());
response = "";
}
finally {
channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes("UTF-8"));

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}
}
}
catch  (Exception e) {
e.printStackTrace();
}
finally {
if (connection != null) {
try {
connection.close();
}
catch (Exception ignore) {}
}
}
}
}

客户端代码稍微涉及:
我们建立一个连接通道,声明一个独家“回调”排队等待回复。
我们订阅“回调”队列,这样我们就可以收到RPC反应。
我们调用方法使实际的RPC请求。
在这里,我们首先生成唯一correlationId号码并将其保存- while循环将使用这个值来捕捉适当的响应。
接下来,我们发布请求消息,有两个属性:replyTo correlationId。
在这一点上我们可以坐下来,等到适当的响应到达。
while循环做一个非常简单的工作,为每一个响应消息它检查如果thecorrelationId是我们正在寻找的人。如果是这样,它节省了响应。
最后我们将响应返回给用户。
客户端请求：

RPCClient fibonacciRpc = new RPCClient();

System.out.println(" [x] Requesting fib(30)");
String response = fibonacciRpc.call("30");
System.out.println(" [.] Got '" + response + "'");

fibonacciRpc.close();

本文提供的设计并不是唯一可能的实现RPC服务,但它有一些重要的优点:
如果RPC服务器太慢了,你可以扩大通过运行另一个。尝试运行第二个RPCServer在一个新的控制台。
在客户端,RPC需要发送和接收消息只有一个。不需要像queueDeclare同步调用。由于RPC客户端只需要一个网络往返一个RPC请求。