远程过程调用
(RPC)
(Java 客户端版)
在第二章（工作队列 Work Queues）中，我们学习了如何使用工作队列在多用户之间分配耗时的任务。
但是如果我们需要在远程电脑上运行一个函数方法，并且还要等待一个返回结果该怎么办？好吧，这将会是一个不一样的故事。这种模式通常被我们称为远程过程调用或者RPC.
在本章教程中，我们将会学习使用RabbitMQ去搭建一个RPC系统：一个客户端和一个可以升级(扩展)的RPC服务器。由于我们没有任何的值得分配的耗时的任务，所以我们将创建一个返回斐波那契数列的虚拟的RPC服务。
客户端接口(Client interface)
为了说明如何使用去使用一个PRC服务，我们将创建一个简单的客户端类。它将暴露一个命名为call的方法，这个方法发送一个阻塞的PRC请求，直到收到一个响应的回复：
FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);
关于RPC的注释：
尽管RPC是一个十分常见的计算机设计模型，但是它经常被批评。当一个开发者不能清晰的认识到请求的函数方法是本地的方法还是一个缓慢的PRC时就会出现问题。这样混乱的结果导致了一个不可预知的系统，并且增加了一些不必要的复杂的调试工作。所以，误用PRC可能导致一些不可维护错综复杂的代码，化简为繁。
铭记这一点，请遵循以下的建议：
- 确认一个函数方法，究竟是本地的还是远程的。
- 记录你的系统，确保组件之间的依赖关系是清晰的。
- 处理错误问题：当RPC服务器长时间宕机时，客户点应该如何做出回应？
为了避免RPC中出现的疑问，如果可以，你应该使用异步管道的方式来代替RPC的阻塞，请求的结果会被异步推送到下一个计算阶段。
回调队列（Callback queue）
通常情况下，使用RabbitMQ去实现RPC很容易。一个客户端发送请求信息，对应得到一个服务器端回复的响应信息。为了得到响应信息，我们需要在请求的时候发送一个“回调”队列地址。我们可以使用默认的(在JAVA客户端中独有)。让我们试试吧：
callbackQueueName = channel.queueDeclare().getQueue();BasicProperties props = new BasicProperties.Builder().replyTo(callbackQueueName).build();
channel.basicPublish("", "rpc_queue", props, message.getBytes());
// ... then code to read a response message from the callback_queue ...
消息属性：
AMQP 0-9-1协议预先定义了发送一个消息需要包含的14个属性。但是其中大多数的属性很少被使用，以下的除外：
1．deliveryMode：将消息标记为持久消息（值为2）或短暂消息（任何其他值）。 从第二个教程可以记住这个属性。
2．contentType：用于描述mime类型的编码。 例如：对于经常使用的JSON编码，将此属性设置为：application / json。
3．replyTo：通常用来命名一个回调队列。
4．correlationId：用于将RPC响应与请求相关联。
我们需要这个新的导入：
import com.rabbitmq.client.AMQP.BasicProperties;
关联id (correlationId)：
在上面提出的方法中，我们建议为每个RPC请求创建一个回调队列。 这是非常低效的，但幸运的是有一个更好的方法:让我们为每个客户端创建一个回调队列。
这就提出了一个新的问题，在队列中得到一个响应时，我们不清楚这个响应所对应的是哪一条请求。这时候就是使用关联id（correlationId）的时候了。我们将为每一条请求设置唯一的的值。稍后，当我们在回调队列里收到一条消息的时候，我们将查看它的属性，我们可以匹配出对应的请求和响应。如果我们发现了一个未知的id值，我们可以安全的丢弃这条消息，因为它不属于我们的请求。
你可以会问，为什么我们在回调队列中忽略未知消息，而不是判断失败呢？这是由于一种可能：在服务器端发生资源竞争(导致产生未知id)。尽管不太现实，很有可能RPC在刚刚发送给我们答案后，但是在发送确认请求消息前挂掉了。如果出现这种情况，重启RPC服务器后，我们将重新处理该条请求。这就是为什么我们要在客户端优雅的处理重复的响应，并且RPC理应上是幂等的。
概要：
http://www.rabbitmq.com/img/tutorials/python-six.png
我们的RPC将像这样工作：
1．当客户端启动时，它创建一个独有的匿名回调队列。
对于一条RPC请求，客户端发送一个具有两个属性的消息：replyTo：包含回调队列和correlationId：包含每个请求的唯一值。
2．请求发送到rpc_queue队列。
3．RPC worker（aka：server）正在等待队列上的请求。 当出现请求时，它将执行该作业，并使用队列中replyTo字段，将结果发送回客户端。
4．客户端等待回调队列中的数据。当一条消息出现时，它会检查correlationId属性，如果和请求中correlationId的值匹配，就返回响应结果给应用程序。
把他们整合到一起：
斐波纳契任务：
private static int fib(int n)
{
if (n == 0) return 0;
if (n == 1) return 1;
return fib(n-1) + fib(n-2);
}
我们声明斐波纳契方法。它只假定输入有效的正整数。(不要期望它执行大数字，并且它可能是用最慢的递归方法实现的。)
我们的RPC服务器RPCServer.java的代码如下所示：
</pre>
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";
public static void main(String[] argv) {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = null;
try {
connection = factory.newConnection();
final Channel channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
channel.basicQos(1);
System.out.println(" [x] Awaiting RPC requests");
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(properties.getCorrelationId())
.build();
String response = "";
try {
String message = new String(body,"UTF-8");
int n = Integer.parseInt(message);
System.out.println(" [.] fib(" + message + ")");
response += fib(n);
}
catch (RuntimeException e){
System.out.println(" [.] " + e.toString());
}
finally {
channel.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
channel.basicAck(envelope.getDeliveryTag(), false);
// RabbitMq consumer worker thread notifies the RPC server owner thread
synchronized(this) {
this.notify();
}
}
}
};
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
// Wait and be prepared to consume the message from RPC client.
while (true) {
synchronized(consumer) {
try {
consumer.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
} catch (IOException | TimeoutException e) {
e.printStackTrace();
} finally {
if (connection != null)
try {
connection.close();
} catch (IOException _ignore) {}
}
}
}
<pre>
服务器代码比较简单：
1．像往常一样，我们开始建立连接，通道并声明队列。
2．我们可能想执行更多的服务器进程。为了在多个服务器上做到负载均衡，我们需要在channel.basicQos中设置prefetchCount设置。
3．我们使用basicConsume访问队列，我们以对象（DefaultConsumer）的形式提供一个回调，该对象将执行该操作并将响应发送回来。
我们的RPC客户端的代码RPCClient.java：
</pre>
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.TimeoutException;
public class RPCClient {
private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName;
public RPCClient() throws IOException, TimeoutException {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();
replyQueueName = channel.queueDeclare().getQueue();
}
public String call(String message) throws IOException, InterruptedException {
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"));
final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
});
return response.take();
}
public void close() throws IOException {
connection.close();
}
//...
<pre>
客户端代码稍微复杂点：
- 我们建立一个连接和通道，并声明一个独有的’回调’队列作为回复。
- 我们订阅’回调’队列，这样我们可以收到RPC响应信息。
- 我们请求方法创建一个实际的RPC请求。
- 在这里，我们首先生成一个唯一的correlationId数字并保存它，我们在DefaultConsumer中实现handleDelivery的时候将使用此值来捕获相应的响应。
- 接下来，我们发布请求消息，包含两个属性：replyTo和correlationId。
- 现在，我们可以坐下来等待适当的响应到达。
- 由于我们消费者交付处理发生在分开的线程中，因此有时候我们需要在响应到达前挂起主线程。使用BlockingQueue是其中的一种解决方案。 这里我们创建ArrayBlockingQueue，容量设置为1，因为我们只需要等待一个响应。
- handleDelivery方法正在做一个非常简单的工作，它检查每个的响应消息的correlationId是否是我们正在寻找的。 如果是，它会把响应消息放入BlockingQueue中。
- 同时主线程正在等待从BlockingQueue取出响应的消息。
最后，我们将响应返回给用户。
创建客户请求：
RPCClient fibonacciRpc = new RPCClient();&amp;nbsp;
System.out.println(" [x] Requesting fib(30)");
String response = fibonacciRpc.call("30");
System.out.println(" [.] Got '" + response + "'");
fibonacciRpc.close();
现在是时候去演示我们完整的实例代码了(RPCClient.java和RPCServer.java)，包括基本异常处理。
像往常一样编译和设置类路径（参见教程一）：
javac -cp $CP RPCClient.java RPCServer.java
我们的RPC服务现在已经准备好了。 我们可以启动服务器：
</pre> java -cp $CP RPCServer<em># => [x] Awaiting RPC requests</em>
请求运行客户端的fibonacci数列：
</pre> java -cp $CP RPCClient<em># => [x] Requesting fib(30)</em>
这里提出的设计不是RPC服务的唯一可能的实现方法，但是其具有一些重要的优点：
- 如果RPC服务器太慢，可以通过运行另一个RPC服务器进行扩展。 尝试在新的控制台中运行第二个RPCServer。
- 在客户端，RPC只需要发送和接收一条消息。 不需要像queueDeclare这样的同步调用。 因此，对于单个RPC请求，RPC客户端只需要一次网络往返。
我们的代码仍然非常简单，并且不会尝试解决更复杂（但是重要的）问题，例如：
- 如果没有服务器运行，客户端应该如何反应？
- 客户端是否应该拥有PRC的一些超时请求的处理？
- 如果服务器发生故障并引发异常，是否应该将其转发给客户端？
- 在执行之前防止无效的传入消息（例如检查边界，类型）。
如果要进行实验，您可能会发现management UI对查看队列很有用。
原创文章，作者：ItWorker，如若转载，请注明出处：https://blog.ytso.com/99539.html