什麼是RPC？

RPC是指遠端過程呼叫，也就是說兩臺伺服器A，B，一個應用部署在A伺服器上，想要呼叫B伺服器上應用提供的函式/方法，由於不在一個記憶體空間，不能直接呼叫，需要通過網路來表達呼叫的語義和傳達呼叫的資料。

為什麼RPC呢？就是無法在一個程序內，甚至一個計算機內通過本地呼叫的方式完成的需求，比如不同的系統間的通訊，甚至不同的組織間的通訊。由於計算能力需要橫向擴充套件，需要在多臺機器組成的叢集上部署應用，

RPC的協議有很多，比如最早的CORBA，Java RMI，Web Service的RPC風格，Hessian，Thrift，甚至Rest API。

RabbitMQ怎麼實現RPC呼叫？

Callback Queue

一般在RabbitMQ中做RPC是很簡單的。客戶端傳送請求訊息，伺服器回覆響應的訊息。為了接受響應的訊息，我們需要在請求訊息中傳送一個回撥佇列。可以使用預設的佇列（which is exclusive in the java client.）:

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

Message properties

AMQP協議為訊息預定義了一組14個屬性。

private String contentType;
private String contentEncoding;
private Map<String,Object> headers;
private Integer deliveryMode;
private Integer
 
 priority;
private String correlationId;
private String replyTo;
private String expiration;
private String messageId;
private Date timestamp;
private String type;
private String userId;
private String appId;
private String clusterId;

大部分的屬性是很少使用的。除了以下幾種（其餘有興趣可以自行檢視）：

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

Correlation Id

在上述方法中為每個RPC請求建立一個回撥佇列。這是很低效的。幸運的是，一個解決方案：可以為每個客戶端建立一個單一的回撥佇列。

新的問題被提出，佇列收到一條回覆訊息，但是不清楚是那條請求的回覆。這是就需要使用correlationId屬性了。我們要為每個請求設定唯一的值。然後，在回撥佇列中獲取訊息，檢視這個屬性，關聯response和request就是基於這個屬性值的。如果我們看到一個未知的correlationId屬性值的訊息，可以放心的無視它——它不是我們傳送的請求。

你可能問道，為什麼要忽略回撥佇列中未知的資訊，而不是當作一個失敗？這是由於在伺服器端競爭條件的導致的。雖然不太可能，但是如果RPC伺服器在傳送給我們結果後，傳送請求反饋前就掛掉了，這有可能會發送未知correlationId屬性值的訊息。如果發生了這種情況，重啟RPC伺服器將會重新處理該請求。這就是為什麼在客戶端必須很好的處理重複響應，RPC應該是冪等的。

Summary

RPC的處理流程：

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

Demo Code

這裡採用官網的一個例子來說明，RPC客戶端通過RPC呼叫伺服器來計算斐波那契額值。
首先是服務端的程式碼：

public class RPCServer {
    private static final String RPC_QUEUE_NAME = "rpc_queue";

    public static void main(String args[]) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitConfig.ip);
        factory.setPort(RabbitConfig.port);
        factory.setUsername(RabbitConfig.username);
        factory.setPassword(RabbitConfig.password);

        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 repsonse = ""+fib(n);
            channel.basicPublish("", props.getReplyTo(), replyProps, repsonse.getBytes());
            channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
        }
    }

    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客戶端：

public class RPCClient {
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";
    private String replyQueueName;
    private QueueingConsumer consumer;

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RabbitConfig.ip);
        factory.setPort(RabbitConfig.port);
        factory.setUsername(RabbitConfig.username);
        factory.setPassword(RabbitConfig.password);

        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 IOException,
            ShutdownSignalException, ConsumerCancelledException,
            InterruptedException {
        String response = null;
        String corrId = 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();
    }

    public static void main(String args[]) throws Exception{
        RPCClient fibRpc = new RPCClient();
        System.out.println(" [x] Requesting fib(30)");
        String response = fibRpc.call("30");
        System.out.println(" [.] Got '"+response+"'");
        fibRpc.close();

    }
}

參考資料

歡迎支援《RabbitMQ實戰指南》以及關注微信公眾號：朱小廝的部落格。