在工作队列一章中，我们学会了如何使用工作队列来处理多个工作进程间分发任务，但如果我们想要运行远程计算机上的函数来获得结果呢？这就是本章要处理的问题RPC。

　　本节我们会使用RabbitMQ构建一个RPC系统：一个客户端和一个可扩展的RPC服务器。因为我们没有任何耗时的任务值得分发下去，我们构建一个虚拟的服务来返回斐波纳契数列。

　　客户端接口

　　我们创建一个客户端类来说明如何使用RPC服务，暴露一个call方法来发送RPC请求和数据获取结果。

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

　　尽管RPC是编程中一种常见的模式，但其也常常饱受批评。因为程序员常常不知道调用的方法是本地方法还是一个RPC方法，这在调试中常常增加一些不必要的复杂性。我们应该简化代码，而不是滥用RPC导致代码变的臃肿。

　　回调队列

　　一般来说，通过RabbitMQ实现RPC非常简单，客户端发送一个请求消息，服务端响应消息就完成了。为了接收到响应内容，我们在请求中发送”callback“队列地址，也可以使用默认的队列。

callbackQueueName = channel.queueDeclare().getQueue();BasicProperties props = new BasicProperties .Builder().replyTo(callbackQueueName) .build(); channel.basicPublish("", "rpc_queue", props, message.getBytes());

   

　　AMQP协议中预定了14个消息属性，除了下面几个，其它的都很少使用：

　　deliveryMode ： 标识消息是持久化还是瞬态的。

　　contentType ： 描述 mime-type的编码类型，如JSON编码为”application/json“。

　　replyTo ： 通常在回调队列中使用。

　　correlationId ： 在请求中关联RPC响应时使用。

　　关联Id（Correlation Id）

　　在前面的方法中，要求在每个RPC请求创建回调队列，这可真是一件繁琐的事情，但幸运的是我们有个好方法-在每个客户端创建一个简单的回调队列。

　　这样问题又来了，队列如何知道这些响应来自哪个请求呢？这时候correlationId就出场了。我们在每个请求中都设置一个唯一的值，这样我们在回调队列中接收消息的时候就能知道是哪个请求发送的。如果收到未知的correlationId，就废弃该消息，因为它不是我们发出的请求。

　　你可能会问，为什么抛弃未知消息而不是抛出错误呢？这是由服务器竞争资源所导致的。尽管这不太可能，试想一下，如果RPC服务器在发送完响应后而在发送应答消息前死掉了，重启RPC服务器会重新发送请求。这就是我们在客户机上优雅地处理重复的反应，RPC应该是等同的。

　　（1）客户端启动，创建一个匿名且唯一的回调队列。

　　（2）对每个RPC请求，客户端发送一个包含replyTo和correlationId两个属性的消息。

　　（3）请求发送到rpc_queue队列。

　　（4）RPC服务在队列中等待请求，当请求出现时，根据replyTo字段使用队列将结果发送到客户端。

　　（5）客户端在回调队列中等待数据。当消息出现时，它会检查correlationId属性，如果该值匹配的话，就会返回响应结果给应用。

　　示例代码

　　RPCServer.java

package com.favccxx.favrabbit;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) {				}			}		}	}}

　　RPCClient.java       

package com.favccxx.favrabbit;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;import java.util.UUID;public class RPCClient {	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 = UUID.randomUUID().toString();		BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();		channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));		while (true) {			QueueingConsumer.Delivery delivery = consumer.nextDelivery();			if (delivery.getProperties().getCorrelationId().equals(corrId)) {				response = new String(delivery.getBody(), "UTF-8");				break;			}		}		return response;	}	public void close() throws Exception {		connection.close();	}	public static void main(String[] argv) {		RPCClient fibonacciRpc = null;		String response = null;		try {			fibonacciRpc = new RPCClient();			System.out.println(" [x] Requesting fib(30)");			response = fibonacciRpc.call("30");			System.out.println(" [.] Got '" + response + "'");		} catch (Exception e) {			e.printStackTrace();		} finally {			if (fibonacciRpc != null) {				try {					fibonacciRpc.close();				} catch (Exception ignore) {				}			}		}	}}

　　先启动RPCServer，然后运行RPCClient,控制台输出如下内容

RPCClient[x] Requesting fib(30) RPCClient[.] Got '832040' RPCServer[x] Awaiting RPC requests RPCServer[.] fib(30)