您的位置：首页 > 其它

转载RabbitMQ入门（6）--远程调用

2015-11-13 11:23 218 查看

远程过程调用(RPC)

(使用Java客户端)

在指南的第二部分，我们学习了如何使用工作队列将耗时的任务分布到多个工作者中。

但是假如我们需要调用远端计算机的函数，等待结果呢？好吧，这又是另一个故事了。这模式通常被称为远程过程调用或RPC。

在这部分，我们将会使用RabbitMQ构建一个RPC系统：一个客户端和一个可扩展的RPC服务器。由于我们还没有值得分散的耗时任务，我们将会创建一个虚拟的RPC服务，用来返回Fibonacci(斐波纳契数列)。

用户接口

为了说明RPC服务如何使用，我们将会创建一个简单德客户端类。它会暴露一个叫

call

的方法，用来发送一个RPC请求，在响应回复之前都会一直阻塞：

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

RPC方面的注意
虽然RPC在电脑运算方面是一个十分普通的模式，但是它依旧常常受批判的。
如果一个程序员没有意识到函数

call

是本地的还是一个迟钝的RPC。这结果是不可预知的很让你困惑的，并且会增加不必要的复杂调试。与简化软件相反，误用RPC会导致不可维护的意大利面条代码（译者注：原文是spaghetti code可能形容代码很长很乱）。

思想中煎熬，考虑下接下来的建议：
确保明显区分哪个是函数

call

是本地调用的，哪个是远端调用的。

给你的系统加上文档，让组件之间的依赖项清晰可见的。

处理错误事件。当RPC服务器很久没有响应了，客户端应该如何响应？
当关于RPC的所有疑问消除，在你可以的情况下，你应该使用一个异步的管道，代替RPC中阻塞，结果会异步的放入接下来的计算平台。

回收队列

一般来说在RabbitMQ上做RPC是容易的。一个客户端发送一个请求消息，一个服务器返回响应消息。为了接受到响应，我们需要再请求中带上一个

callback

队列的地址。我们可以使用默认队列(那个在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协议预先确定了消息中的14个属性。他们大多数属性很少使用，除了下面这些例外：

deliveryMode

:将一个消息标记为持久化(值为2)或者瞬态的(其他值)。你可能从第二部分中记起这个属性。

contentType

:用来描述媒体类型的编码。例如常常使用的JSON编码，这是一个好的惯例，设置这个属性为：

application/json

。

replyTo

:通常来命名回收队列的名字。

correlationId

:对RPC加速响应请求是很有用的。

我们需要这个新的引用：

import com.rabbitmq.client.AMQP.BasicProperties;

摘要

我们的RPC将会像这样工作：

当客户端启动，它会创建一个匿名的独占的回收队列。
对于一个RPC请求，客户端会发送一个消息中有两个属性：

replyTo

,要发送的的回收队列和

correlationId

,对于每一个请求都是唯一值。
这请求发送到

rpc_queue

队列中。
这RPC工作者(亦称：服务器)等候队列中的请求。当请求出现，它处理这工作并发送携带结果的信息到客户端，使用的队列是消息属性

replTo

中的那个。
客户端等待回收队列中的数据。当一个消息出现，它会检查

correlationId

属性。如果它符合请求中的值，它会返回这响应给应用程序。

把所有的放在一起

斐波那契任务：

private static int fib(int n) throws Exception {
if (n == 0) return 0;
if (n == 1) return 1;
return fib(n-1) + fib(n-2);
}

我们声明我们的斐波那契函数。它假定一个合法的正整数做为输入参数。（不要期望这个可以处理大量数字，它可能是最慢的递归实现了）。
我们的RPC服务器

RPCServer.java

的代码：

private static final String RPC_QUEUE_NAME = "rpc_queue";

ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");

Connection connection = factory.newConnection();
Channel 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) {
QueueingConsumer.Delivery delivery = consumer.nextDelivery();

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

String message = new String(delivery.getBody());
int n = Integer.parseInt(message);

System.out.println(" [.] fib(" + message + ")");
String response = "" + fib(n);

channel.basicPublish( "", props.getReplyTo(), replyProps, response.getBytes());

channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
}

这服务器代码是相当简单明了的：
如往常一样，我们开始建立连接，通道和声明队列。
我们可能想运行不止一个服务器进程。为了均衡的负载到多个服务器上，我们需要设置

channel.basicQos

中的

prefetchCount

属性。
我们使用

basicConsume

访问队列。然后进入

while

循环，我们等待请求消息，处理工作，发送响应。

我们RPC客户端

RPCClient.java

的代码：

private Connection connection;
private Channel channel;
private String requestQueueName = "rpc_queue";
private String replyQueueName;
private QueueingConsumer consumer;

public RPCClient() throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
connection = factory.newConnection();
channel = connection.createChannel();

replyQueueName = channel.queueDeclare().getQueue();
consumer = new QueueingConsumer(channel);
channel.basicConsume(replyQueueName, true, consumer);
}

public String call(String message) throws Exception {
String response = null;
String corrId = java.util.UUID.randomUUID().toString();

BasicProperties props = new BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();

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

while (true) {
QueueingConsumer.Delivery delivery =consumer.nextDelivery();
if (delivery.getProperties().getCorrelationId().equals(corrId)) {
response = new String(delivery.getBody());
break;
}
}

return response;
}

public void close() throws Exception {
connection.close();
}

The client code is slightly more involved:
这客户端代码是更加清晰：
我们建立一个连接和通道并且声明一个独占的

callback

队列用来等待答复。
我们订阅这个

callback

队列，以便于我们可以接收到RPC响应。
我们的

call

方法做这真正的RPC请求。
接着，我们首次生成一个唯一的

correlationId

数字并且保存它，在循环中使用这个值找到合适的响应。
接下来，我们发布请求消息，带着两个属性：

replyTo

和

correlationId

。
这时候，我们可以坐下来，等着合适的响应抵达。
这循环中做了个简单德工作，检查每一个响应消息中

correlationId

值，是否是它要寻找的。如果是，它会保存这响应。
最终，我们把响应返回给用户。

制造客户端请求：

RPCClient fibonacciRpc = new RPCClient();

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 rabbitmq-client.jar RPCClient.java RPCServer.java

我们的RPC服务现在准备好了，我们启动着服务器：

$ java -cp $CP RPCServer
[x] Awaiting RPC requests

为了请求一个斐波那契数字，运行客户端：

$ java -cp $CP RPCClient
[x] Requesting fib(30)

现在的设计不仅仅可以实现一个RPC服务，并且它还有几项重要的优势：
如果RPC服务器反应太迟缓，你可以通过运行另一个程序来扩展。试着通过一个新的控制平台来运行第二个RPC服务器。在客户端这边，RPC要求仅发送和接收一个消息。像

queueDeclare

非同步调用是被要求的。因此，RPC客户端仅仅需要一个网络循环的单一RPC请求。

我们的代码一直是十分简单的，不能试着解决更复杂（但是重要）的问题，比如：
如果没有服务器运行，客户端如何响应？
客户端是否对RPC的超时有处理？
如果服务器发生故障，抛出一个异常，是否应该传递到客户端？
在处理之前把进入来的非法消息隔离掉（检查界限，类型）。

内容来自用户分享和网络整理，不保证内容的准确性，如有侵权内容，可联系管理员处理

标签：

相关文章推荐

新的分享

章节导航