RabbitMQ (六) 建構RPC

但是付過我們需要在遠端電腦上運作一個方法然後等待結果，該怎麼辦？這是不同的需求。這個模式通常叫做RPC。

本文我們将使用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（回調隊列）

一般在RabbitMQ中做RPC是很簡單的。用戶端發送請求消息，伺服器回複響應的消息。為了接受響應的消息，我們需要在請求消息中發送一個回調隊列。可以用預設的隊列（僅限java用戶端）。試一試吧：

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個屬性。大部分的屬性是很少使用的。除了一下幾種：

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（總結）

RabbitMQ (六) 建構RPCRabbitMQ (六) 建構RPCClient interface（用戶端接口）Callback queue（回調隊列）Message properties（消息屬性）Correlation IdSummary（總結）完整的執行個體

我們的RPC的處理流程：

當用戶端啟動時，建立一個匿名的回調隊列。
用戶端為RPC請求設定2個屬性：replyTo，設定回調隊列名字；correlationId，标記request。
請求被發送到rpc_queue隊列中。
RPC伺服器端監聽rpc_queue隊列中的請求，當請求到來時，伺服器端會處理并且把帶有結果的消息發送給用戶端。接收的隊列就是replyTo設定的回調隊列。
用戶端監聽回調隊列，當有消息時，檢查correlationId屬性，如果與request中比對，那就是結果了。

完整的執行個體

RPC伺服器端（RPCServer.java）

import com.rabbitmq.client.*;

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


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

    /**
     * 斐波那契
     * @param n
     * @return
     */
    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[] args) throws IOException, TimeoutException
    {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("120.79.222.144");

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

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

        //  清空一個隊列
        channel.queuePurge(RPC_QUEUE_NAME);

        //這告訴RabbitMQ一次不同時向一個worker發送一條消息（公平調遣）
        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()
                    // 用于将RPC響應與請求相關聯。
                    .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"));

                // basicAck 确認是否接受一條或者多條消息
                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();
                }
            }
        }
    }
}

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
{
    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";

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

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

    public static void main(String[] args)
    {
        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 {
        // 生成corrId
        final String corrId = UUID.randomUUID().toString();

        //為每一個用戶端擷取一個随機的回調隊列
        String replyQueueName = channel.queueDeclare().getQueue();
        //設定replyTo和correlationId屬性值
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        //發送消息到rpc_queue隊列
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        final BlockingQueue<String> 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();
    }
}

RabbitMQ (六) 建構RPCRabbitMQ (六) 建構RPCClient interface（用戶端接口）Callback queue（回調隊列）Message properties（消息屬性）Correlation IdSummary（總結）完整的執行個體

RabbitMQ (六) 建構RPC

Client interface（用戶端接口）

Callback queue（回調隊列）

Message properties（消息屬性）

Correlation Id

Summary（總結）

完整的執行個體

繼續閱讀

RabbitMQ：MessageListenerAdapter消息監聽器擴充卡使用與源碼分析

CentOS：下載下傳RabbitMQ并遠端通路

RabbitMQ：交換機（topic exchange）

RabbitMQ：@RabbitListener注解簡化消息監聽

RabbitMQ：Docker Compose部署RabbitMQ叢集

RabbitMQ：交換機（direct exchange）

RabbitMQ：消費端限流

RabbitMQ安裝及配置前言介紹Erlang安裝RabbitMQ安裝

RabbitMQ：消費端自定義監聽

RabbitMQ：整合Spring Boot

RabbitMQ：MessageConverter消息轉換器

rabbitmq連接配接出錯，記錄一下

RabbitMQ：交換機（fanout exchange）

rabbitmq 指令行操作

Rabbitmq指令行手動建立隊列

RabbitMQ指令行的基本操作