摘要：如果涉及返回值，就要用到本章提到的了。方法發(fā)送請求，并阻塞知道結(jié)果返回。當(dāng)有消息時(shí)，進(jìn)行計(jì)算并通過指定的發(fā)送給客戶端。當(dāng)接收到，則檢查。如果和之前的匹配，則將消息返回給應(yīng)用進(jìn)行處理。

在第二章中我們學(xué)習(xí)了如何使用Work模式在多個(gè)worker之間派發(fā)時(shí)間敏感的任務(wù)。這種情況是不涉及到返回值的，worker執(zhí)行任務(wù)就好。如果涉及返回值，就要用到本章提到的RPC（Remote Procedure Call）了。

本章我們使用RabbitMQ來構(gòu)建一個(gè)RPC系統(tǒng)：一個(gè)客戶端和一個(gè)可擴(kuò)展的RPC服務(wù)端。我們讓RPC服務(wù)返回一個(gè)斐波那契數(shù)組。

我們創(chuàng)建一個(gè)簡單的客戶端類來演示如何使用RPC服務(wù)。call方法發(fā)送RPC請求，并阻塞知道結(jié)果返回。

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

RPC貼士
雖然RPC的使用在計(jì)算機(jī)領(lǐng)域非常普遍，但是卻經(jīng)常受到批評。主要問題是編碼人如果不注意使用的方法是本地還是遠(yuǎn)程時(shí)，往往會造成問題。往往讓系統(tǒng)變得不可預(yù)知，增加不必要的復(fù)雜性和調(diào)試的難度。對此我們有如下幾點(diǎn)建議：

是本地方法還是遠(yuǎn)程方法要一目了然
把系統(tǒng)的依賴寫進(jìn)文檔
系統(tǒng)要處理好超時(shí)的問題

如果可以盡量使用異步的pipeline來替代像RPC這種阻塞的操作。

在RabbitMQ上實(shí)現(xiàn)RPC是非常簡單的?？蛻舳税l(fā)送一個(gè)request message，服務(wù)端回應(yīng)一個(gè)response message。為了接受response message我們需要在發(fā)送request message的時(shí)候附帶上"callback" queue的地址。我們可以使用默認(rèn)的queue。

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的屬性
AMQP 0-9-1協(xié)議預(yù)定義了14個(gè)消息屬性，其中大部分很少使用，下面的屬性較為常用

deliverMode: 標(biāo)記message為持久(設(shè)置為2)或其他值。
contentType:message的編碼類型，我們經(jīng)常使用JSON編碼，則設(shè)置為application/json
replyTo: 命名回調(diào)queue
correlationId:將RPC的請求和回應(yīng)關(guān)聯(lián)起來

需要引入新的類

import com.rabbitmq.client.AMQP.BasicProperties;

在上面的代碼中，每次RPC請求都會創(chuàng)建一個(gè)用于回調(diào)的臨時(shí)queue，我們有更好的方法，我們?yōu)槊恳粋€(gè)client創(chuàng)建一個(gè)回調(diào)queue。

但是這樣有新的問題，從回調(diào)queue中收到response無法和相應(yīng)的request關(guān)聯(lián)起來。這時(shí)候就是correlationId屬性發(fā)揮作用的時(shí)候了。為每個(gè)request中設(shè)置唯一的值，在稍后的回調(diào)queue中收到的response里也有這個(gè)屬性，基于此，我們就可以關(guān)聯(lián)之前的request了。如果我們遇到一個(gè)匹配不到的correlationId,那么丟棄的行為是安全的。

你可能會問，為什么我們忽略這些無法匹配的message，而不是當(dāng)做一個(gè)錯(cuò)誤處理呢？主要是考慮服務(wù)端的競態(tài)條件，如果RPC服務(wù)器在發(fā)送response之后就宕機(jī)了，但是卻沒有發(fā)送ack消息。那么當(dāng)RPC Server重啟之后，會繼續(xù)執(zhí)行這個(gè)request。這就是為什么client需要冪等處理response。

我們的RPC向下面這樣進(jìn)行工作：

對于一個(gè)RPC request，客戶端發(fā)送message時(shí)設(shè)置兩個(gè)屬性：replyTo設(shè)置成一個(gè)沒有名字的request獨(dú)有的queue；為每個(gè)request設(shè)置一個(gè)唯一的correlationId。

request發(fā)送到rpc_queue

RPC worker監(jiān)聽rpc_queue。當(dāng)有消息時(shí)，進(jìn)行計(jì)算并通過replyTo指定的queue發(fā)送message給客戶端。

客戶端監(jiān)聽回調(diào)queue。當(dāng)接收到message，則檢查correlationId。如果和之前的request匹配，則將消息返回給應(yīng)用進(jìn)行處理。

斐波那契處理函數(shù)

private static int fib(int n) {
    if (n == 0) return 0;
    if (n == 1) return 1;
    return fib(n-1) + fib(n-2);
}

這是一個(gè)簡易的實(shí)現(xiàn)，如果傳入一個(gè)較大的值，將會是個(gè)災(zāi)難。
RPC服務(wù)器的代碼為RPCServer.java, 代碼是很簡單明確的

先是建立connection，channel和聲明queue.

設(shè)置prefetchCount，我們基于請求頻繁程度，會啟動(dòng)多個(gè)RPC Server

使用basicConsume來接收，該方法提供回調(diào)參數(shù)設(shè)置（DeliverCallback）.

RPC客戶端的代碼為RPCClient.java,代碼略微有點(diǎn)復(fù)雜

建立connection和channel。

call方法來發(fā)送RPC請求

生成correlationId

生成默認(rèn)名字的queue用于reply，并訂閱它

發(fā)送request message，設(shè)置參數(shù)replyTo和correlationId.

然后返回并開始等待response到達(dá)

因?yàn)橄M(fèi)者發(fā)送response是在另一個(gè)線程中，我們需要讓main線程阻塞，在這里我們使用BlockingQueue。

消費(fèi)者進(jìn)行簡單的處理，為每一個(gè)response message檢查其correlationId,如果是，則將response添加進(jìn)阻塞隊(duì)列

main函數(shù)阻塞在BlockingQueue返回

將response返回給用戶

RPCClient.java完整代碼

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

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 implements AutoCloseable {

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

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got "" + response + """);
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    public String call(String message) throws IOException, InterruptedException {
        final String corrId = UUID.randomUUID().toString();

        String replyQueueName = channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        final BlockingQueue response = new ArrayBlockingQueue<>(1);

        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                response.offer(new String(delivery.getBody(), "UTF-8"));
            }
        }, consumerTag -> {
        });

        String result = response.take();
        channel.basicCancel(ctag);
        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
}

RPCServer.java完整代碼

import com.rabbitmq.client.*;

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) throws Exception {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("localhost");

        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
            channel.queuePurge(RPC_QUEUE_NAME);

            channel.basicQos(1);

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

            Object monitor = new Object();
            DeliverCallback deliverCallback = (consumerTag, delivery) -> {
                AMQP.BasicProperties replyProps = new AMQP.BasicProperties
                        .Builder()
                        .correlationId(delivery.getProperties().getCorrelationId())
                        .build();

                String response = "";

                try {
                    String message = new String(delivery.getBody(), "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("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
                    channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
                    // RabbitMq consumer worker thread notifies the RPC server owner thread
                    synchronized (monitor) {
                        monitor.notify();
                    }
                }
            };

            channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));
            // Wait and be prepared to consume the message from RPC client.
            while (true) {
                synchronized (monitor) {
                    try {
                        monitor.wait();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            }
        }
    }
}