假如我們想要呼叫遠端的一個方法或函式並等待執行結果，也就是我們通常說的遠端過程呼叫（Remote Procedure Call）。怎麼辦？

今天我們就用RabbitMQ來實現一個簡單的RPC系統：客戶端傳送一個請求訊息，服務端以一個響應訊息迴應。為了能夠接收到響應，客戶端在傳送訊息的同時傳送一個回撥佇列用來告訴服務端響應訊息傳送到哪個佇列裡面。也就是說每個訊息一個回撥佇列，在此基礎上我們變下，將回調佇列定義成類的屬性，這個每個客戶端一個佇列，同一個客戶端的請求共用一個佇列。那麼接下來有個問題，怎麼知道這個佇列裡面的響應訊息是屬於哪個佇列的呢？

我們會用到關聯標識（correlationId），每個請求我們都會生成一個唯一的值作為correlationId，這樣每次有響應訊息來的時候，我們就去看correlationId來確定到底是哪個請求的響應訊息，將請求和響應關聯起來。如果收到一個不知道的correlationId，就可以確定不是這個客戶端的請求的響應，可以直接丟棄掉。

一、工作模型

1. 客戶端傳送啟動後，會建立獨特的回撥佇列。對於一個請求傳送配置了兩個屬性的訊息：一個是回撥佇列（圖中的replay_to），一個是correlation。 這個請求會發送到rpc_queue佇列，然後到達服務端處理。
2. 服務端等待rpc_queue佇列的請求。當有請求到來時，它就會開始幹活並將結果通過傳送訊息來返回，該返回訊息傳送到replyTo指定的佇列。
3. 客戶端將等待回撥佇列返回資料。當返回的訊息到達時，它將檢查correlation id屬性。如果該屬性值和請求匹配，就將響應返回給程式。

二、程式碼實現

接下來看程式碼實現：

1.  客戶端

   public
    
    class RpcClient {
   
       Connection connection = null;
       Channel channel = null;
       //回撥佇列:用來接收服務端的響應訊息
       String queueName = "";
   
       // 定義RpcClient
       public RpcClient() throws IOException, TimeoutException {
           ConnectionFactory factory = new ConnectionFactory();
           factory.setHost(
    
   "localhost");
           connection = factory.newConnection();
           channel = connection.createChannel();
           queueName = channel.queueDeclare().getQueue();
       }
   
       // 真正的處理邏輯
       public String call(String msg) throws IOException, InterruptedException {
           final String uuid = UUID.randomUUID().toString();
           //後續，服務端根據"replyTo"來指定將返回資訊寫入到哪個佇列
           //後續，服務端根據關聯標識"correlationId"來指定返回的響應是哪個請求的
           AMQP.BasicProperties prop = new AMQP.BasicProperties().builder().replyTo(queueName).correlationId(uuid).build();
   
           channel.basicPublish("", RpcServer.QUEUE_NAME, prop, msg.getBytes());
           final BlockingQueue<String> blockQueue = new ArrayBlockingQueue<String>(1);
           channel.basicConsume(queueName, true, new DefaultConsumer(channel) {
               @Override
               public void handleDelivery(String consumerTag, Envelope envelope,
                       com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) throws IOException {
   
                   if (properties.getCorrelationId().equals(uuid)) {
                       String msg = new String(body, "UTF-8");
   
                       blockQueue.offer(msg);
                       System.out.println("**** rpc client reciver response :[" + msg + "]");
                   }
               }
   
           });
   
           return blockQueue.take();
       }
   
       //關閉連線
       public void close() throws IOException {
           connection.close();
       }
   
       public static void main(String[] args) throws IOException, TimeoutException, InterruptedException {
           RpcClient client = new RpcClient();
           client.call("4");
           client.close();
       }
   }

   傳送請求的時候，它是生產者；接受響應的時候，它是消費者。

2. 服務端

   public class RpcServer {
   
       //RPC佇列名
       public static final String QUEUE_NAME = "rpc_queue";
   
       //斐波那契數列，用來模擬工作任務
       public static int fib(int num) {
           if (num == 0)
               return 0;
           if (num == 1)
               return 1;
           return fib(num - 1) + fib(num - 2);
       }
   
       public static void main(String[] args) throws InterruptedException {
   
           ConnectionFactory factory = new ConnectionFactory();
           factory.setHost("localhost");
           Connection connection = null;
           try {
               // 1.connection & channel
               connection = factory.newConnection();
               final Channel channel = connection.createChannel();
   
               // 2.declare queue
               channel.queueDeclare(QUEUE_NAME, false, false, false, null);
   
               System.out.println("****** rpc server waiting for client request ......");
   
               // 3.每次只接收一個訊息（任務）
               channel.basicQos(1);
               //4.獲取消費例項
               Consumer consumer = new DefaultConsumer(channel) {
                   @Override
                   public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties,
                           byte[] body) throws IOException {
                       BasicProperties prop = new BasicProperties().builder().correlationId(properties.getCorrelationId())
                               .build();
                       String resp = "";
                       try {
                           String msg = new String(body, "UTF-8");
                           resp = fib(Integer.valueOf(msg)) + "";
                           System.out.println("*** will response to rpc client :" + resp);
                       } catch (Exception ex) {
                           ex.printStackTrace();
                       } finally {
                           channel.basicPublish("", properties.getReplyTo(), prop, resp.getBytes());
                           channel.basicAck(envelope.getDeliveryTag(), false);
                       }
   
                   }
               };
               // 5.消費訊息（處理任務）
               channel.basicConsume(QUEUE_NAME, false, consumer);
           } catch (IOException e) {
               e.printStackTrace();
           } catch (TimeoutException e) {
               e.printStackTrace();
           }
       }
   
   }

   接受請求的時候，它是消費者；傳送響應的時候，它是生產者。

3. 執行服務端，開始等待請求
4. 然後執行客戶端，控制檯log:

   服務端（多了一條列印）：
   ****** rpc server waiting for client request ......
   *** will response to rpc client :3
   
   客戶端：
   **** rpc client reciver response :[3]

三、小插曲

剛開始我在寫demo的時候，client中沒有用到阻塞佇列final BlockingQueue<String> blockQueue = new ArrayBlockingQueue<String>(1);，而是直接這樣寫：

@Override
            public void handleDelivery(String consumerTag, Envelope envelope,
                    com.rabbitmq.client.AMQP.BasicProperties properties, byte[] body) throws IOException {

                if (properties.getCorrelationId().equals(uuid)) {
                    String msg = new String(body, "UTF-8");

                    //blockQueue.offer(msg);
                    System.out.println("**** rpc client reciver response :[" + msg + "]");
                }
            }

期望能打印出結果來，但是執行後發現並沒有列印：**** rpc client reciver response :[" + msg + "]的值。

原因是handleDelivery()這個方法是在子執行緒中執行的，這個子執行緒執行的時候，主執行緒會繼續往後執行直到執行了client.close();方法而結束了。

由於主執行緒終止了，導致沒有打印出結果。加了阻塞佇列之後將主執行緒阻塞不執行close()方法，問題就解決了。