如果我們需要在遠端計算機上執行一個函式並等待結果，這種模式通常被稱為遠端過程呼叫或RPC。

在本教程中，我們將使用RabbitMQ構建一個RPC系統：一個客戶端和一個RPC伺服器。我們將建立一個返回斐波那契數字的模擬RPC服務。

整個過程示意圖如下：

客戶端將請求傳送至rpc_queue（我們定義的訊息佇列），然後等待響應；服務端獲取請求，並處理請求，然後將請求結果返回給佇列，客戶端得知請求被響應後獲取結果。

在結果被響應之前，客戶端是被阻塞的，主執行緒會等待RPC響應

如果每個RPC請求都建立一個回撥佇列。這是非常低效，我們建立一個單一的客戶端回撥佇列。

這引發了一個新的問題，在該佇列中收到回覆時，不清楚回覆屬於哪個請求。這就需要用到 correlationId屬性。我們為沒有請求設定唯一的correlationId值。然後，當我們在回撥佇列中收到一條訊息時，我們將獲取這個值，將響應與請求的進行correlationId匹配。如果我們一致就是我們需要的結果，否則就不是。

客戶端代RPCClient 碼如下：

package com.adtec.rabbitmq;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import
 
 java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;

public class RPCClient {

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

    public RPCClient() throws IOException, TimeoutException {
        //建立一個連線和一個通道，併為回撥宣告一個唯一的'回撥'佇列
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        connection = factory.newConnection();
        channel = connection.createChannel();
        //定義一個臨時變數的接受佇列名    
        replyQueueName = channel.queueDeclare().getQueue();
    }
    //傳送RPC請求  
    public String call(String message) throws IOException, InterruptedException {
         //生成一個唯一的字串作為回撥佇列的編號
        String corrId = UUID.randomUUID().toString();
        //傳送請求訊息，訊息使用了兩個屬性：replyto和correlationId
        //服務端根據replyto返回結果，客戶端根據correlationId判斷響應是不是給自己的
        AMQP.BasicProperties props = new AMQP.BasicProperties.Builder().correlationId(corrId).replyTo(replyQueueName)
                .build();

        //釋出一個訊息，requestQueueName路由規則
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        //由於我們的消費者交易處理是在單獨的執行緒中進行的，因此我們需要在響應到達之前暫停主執行緒。
        //這裡我們建立的 容量為1的阻塞佇列ArrayBlockingQueue，因為我們只需要等待一個響應。
        final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);

        // String basicConsume(String queue, boolean autoAck, Consumer callback)
        channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                    byte[] body) throws IOException {
                //檢查它的correlationId是否是我們所要找的那個
                if (properties.getCorrelationId().equals(corrId)) {
                    //如果是，則響應BlockingQueue
                    response.offer(new String(body, "UTF-8"));
                }
            }
        });

        return response.take();
    }

    public void close() throws IOException {
        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 (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        } finally {
            if (fibonacciRpc != null) {
                try {
                    fibonacciRpc.close();
                } catch (IOException _ignore) {
                }
            }
        }
    }
}

服務端代RPCServer 碼如下：

package rabbitmq;

import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

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) {
         //建立連線、通道，並宣告佇列 
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        Connection connection = null;
        try {
            connection = factory.newConnection();
            final Channel channel = connection.createChannel();

            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);

            channel.basicQos(1);

            System.out.println(" [x] Awaiting RPC requests");

            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties,
                        byte[] body) throws IOException {
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                            .correlationId(properties.getCorrelationId()).build();

                    String response = "";

                    try {
                        String message = new String(body, "UTF-8");
                        int n = Integer.parseInt(message);

                        System.out.println(" [.] fib(" + message + ")");
                        response += fib(n);
                    } catch (RuntimeException e) {
                        System.out.println(" [.] " + e.toString());
                    } finally {
                        // 返回處理結果佇列
                        channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
                        //  確認訊息，已經收到後面引數 multiple：是否批量.true:將一次性確認所有小於envelope.getDeliveryTag()的訊息。
                        channel.basicAck(envelope.getDeliveryTag(), false);
                        // RabbitMq consumer worker thread notifies the RPC
                        // server owner thread
                        synchronized (this) {
                            this.notify();
                        }
                    }
                }
            };
            //取消自動確認
            boolean autoAck = false ;
            channel.basicConsume(RPC_QUEUE_NAME, autoAck, consumer);
            // Wait and be prepared to consume the message from RPC client.
            while (true) {
                synchronized (consumer) {
                    try {
                        consumer.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        } catch (IOException | TimeoutException e) {
            e.printStackTrace();
        } finally {
            if (connection != null)
                try {
                    connection.close();
                } catch (IOException _ignore) {
                }
        }
    }
}

測試時先執行服務端，再執行客戶端

為了方便觀察結果，最好將客戶端和服務端在不同workspace實現

客戶端結果