本文我們將使用RabbitMQ構建一個RPC系統：一個客戶端和一個可擴充套件的RPC伺服器端。由於我們沒有任何真實的耗時任務需要分配，所以我們將建立一個虛擬的RPC服務，可以返回斐波納契數列。

Client interface（客戶端介面）

為了說明RPC服務可以使用，我們建立一個簡單的客戶端類。暴露一個方法——傳送RPC請求，然後阻塞直到獲得結果。

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

Callback queue（回撥佇列）

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 ...
 

Message properties（訊息屬性）

• • deliveryMode：標記訊息傳遞模式，2-訊息持久化，其他值-瞬態。在第二篇文章中還提到過。
  • contentType：內容型別，用於描述編碼的mime-type。例如經常為該屬性設定JSON編碼。
  • replyTo：應答，通用的回撥佇列名稱
  • correlationId：關聯ID，方便RPC響應與請求關聯

import com.rabbitmq.client.AMQP.BasicProperties;

Correlation Id

       在上述方法中為每個RPC請求建立一個回撥佇列。這是很低效的。幸運的是，一個解決方案：可以為每個客戶端建立一個單一的回撥佇列。
       新的問題被提出，佇列收到一條回覆訊息，但是不清楚是那條請求的回覆。這是就需要使用correlationId屬性了。我們要為每個請求設定唯一的值。然後，在回撥佇列中獲取訊息，看看這個屬性，關聯response和request就是基於這個屬性值的。如果我們看到一個未知的correlationId屬性值的訊息，可以放心的無視它——它不是我們傳送的請求。
       你可能問道，為什麼要忽略回撥佇列中未知的資訊，而不是當作一個失敗？這是由於在伺服器端競爭條件的導致的。雖然不太可能，但是如果RPC伺服器在傳送給我們結果後，傳送請求反饋前就掛掉了，這有可能會發送未知correlationId屬性值的訊息。如果發生了這種情況，重啟RPC伺服器將會重新處理該請求。這就是為什麼在客戶端必須很好的處理重複響應，RPC應該是冪等的。

Summary（總結）

       我們的RPC的處理流程：

1. 1. 當客戶端啟動時，建立一個匿名的回撥佇列。
   2. 客戶端為RPC請求設定2個屬性：replyTo，設定回撥佇列名字；correlationId，標記request。
   3. 請求被髮送到rpc_queue佇列中。
   4. RPC伺服器端監聽rpc_queue佇列中的請求，當請求到來時，伺服器端會處理並且把帶有結果的訊息傳送給客戶端。接收的佇列就是replyTo設定的回撥佇列。
   5. 客戶端監聽回撥佇列，當有訊息時，檢查correlationId屬性，如果與request中匹配，那就是結果了。

完整的例項

RPC伺服器端（RPCServer.java）

/**
 * RPC伺服器端
 * 
 * @author arron
 * @date 2015年9月30日 下午3:49:01
 * @version 1.0
 */
public class RPCServer {
	
	private static final String RPC_QUEUE_NAME = "rpc_queue";

	public static void main(String[] args) throws Exception {
		ConnectionFactory factory = new ConnectionFactory();
		// 設定MabbitMQ所在主機ip或者主機名
		factory.setHost("127.0.0.1");
		// 建立一個連線
		Connection connection = factory.newConnection();
		// 建立一個頻道
		Channel channel = connection.createChannel();

		//宣告佇列
		channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

		//限制：每次最多給一個消費者傳送1條訊息
		channel.basicQos(1);

		//為rpc_queue佇列建立消費者，用於處理請求
		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();

			//獲取請求中的correlationId屬性值，並將其設定到結果訊息的correlationId屬性中
			BasicProperties props = delivery.getProperties();
			BasicProperties replyProps = new BasicProperties.Builder().correlationId(props.getCorrelationId()).build();
			//獲取回撥佇列名字
			String callQueueName = props.getReplyTo();
			
			String message = new String(delivery.getBody(),"UTF-8");
			
			System.out.println(" [.] fib(" + message + ")");
			
			//獲取結果
			String response = "" + fib(Integer.parseInt(message));
			//先發送回調結果
			channel.basicPublish("", callQueueName, replyProps,response.getBytes());
			//後手動傳送訊息反饋
			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
		}
	}

	/**
	 * 計算斐波列其數列的第n項
	 * 
	 * @param n
	 * @return
	 * @throws Exception
	 */
	private static int fib(int n) throws Exception {
		if (n < 0)
			throw new Exception("引數錯誤，n必須大於等於0");
		if (n == 0)
			return 0;
		if (n == 1)
			return 1;
		return fib(n - 1) + fib(n - 2);
	}
}

/** 
 * 
 * @author arron
 * @date 2015年9月30日 下午3:44:43 
 * @version 1.0 
 */
public class RPCClient {

	private static final String RPC_QUEUE_NAME = "rpc_queue";
	
	private Connection connection;
	private Channel channel;
	private String replyQueueName;
	private QueueingConsumer consumer;
	
	public RPCClient() throws Exception {
	    ConnectionFactory factory = new ConnectionFactory();
		// 設定MabbitMQ所在主機ip或者主機名
		factory.setHost("127.0.0.1");
		// 建立一個連線
		connection = factory.newConnection();
		// 建立一個頻道
		channel = connection.createChannel();
		
		//宣告佇列
		channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
		
		//為每一個客戶端獲取一個隨機的回撥佇列
	    replyQueueName = channel.queueDeclare().getQueue();
	    //為每一個客戶端建立一個消費者（用於監聽回撥佇列，獲取結果）
	    consumer = new QueueingConsumer(channel);
	    //消費者與佇列關聯
	    channel.basicConsume(replyQueueName, true, consumer);
	}
	
	/**
	 * 獲取斐波列其數列的值
	 * 
	 * @param message
	 * @return
	 * @throws Exception
	 */
	public String call(String message) throws Exception{
		String response = null;
	    String corrId = java.util.UUID.randomUUID().toString();

	    //設定replyTo和correlationId屬性值
	    BasicProperties props = new BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName).build();

	    //傳送訊息到rpc_queue佇列
	    channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes());

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

	    return response; 
	}

	public static void main(String[] args) throws Exception {
		RPCClient fibonacciRpc = new RPCClient();   
		String result = fibonacciRpc.call("4");
		System.out.println( "fib(4) is " + result);
	}
}

fib(4) is 3