Remote procedure call (RPC)
(using the Java client)
Prerequisites
This tutorial assumes RabbitMQ is installed and running on localhost on standard port (5672). In case you use a different host, port or credentials, connections settings would require adjusting.
Where to get help
If you're having trouble going through this tutorial you can contact us through the mailing list.
In the second tutorial we learned how to use Work Queues to distribute time-consuming tasks among multiple workers.
在第二個教程中，我們學習了如何使用工作隊列在多個從業人員之間配置設定耗時的任務。
But what if we need to run a function on a remote computer and wait for the result? Well, that's a different story. This pattern is commonly known as Remote Procedure Call or RPC.
但是如果我們需要在遠端計算機上運作功能并等待結果怎麼辦？ 此模式通常稱為遠端過程調用或RPC。
In this tutorial we're going to use RabbitMQ to build an RPC system: a client and a scalable RPC server. As we don't have any time-consuming tasks that are worth distributing, we're going to create a dummy RPC service that returns Fibonacci numbers.
在本教程中，我們将使用RabbitMQ建構一個RPC系統：它包含一個用戶端和一個可擴充的RPC伺服器。 由于我們沒有任何值得分發的耗時任務，我們将建立一個傳回斐波納契數字的虛拟RPC服務來代替。
1.Client interface 用戶端接口
To illustrate how an RPC service could be used we're going to create a simple client class. It's going to expose a method named call which sends an RPC request and blocks until the answer is received:
為了說明如何使用RPC服務，我們将建立一個簡單的用戶端類。 它将公開一個名為call的方法，該方法發送RPC請求并阻塞，直到接收到響應：
FibonacciRpcClient fibonacciRpc = new FibonacciRpcClient();
String result = fibonacciRpc.call("4");
System.out.println( "fib(4) is " + result);
A note on RPC
雖然RPC是一個很常見的計算模式，但是它并不總是是一個很好的解決方式。當程式員不知道函數調用是本地完成還是通過緩慢的RPC時，可能會出現問題。 帶來未知的混亂，還對于調試增加了的不必要的複雜性。 而沒有達到不是簡化軟體的目的，是以濫用RPC可能導緻代碼不可維護。
銘記這一點，請考慮以下建議：
-
確定哪個函是在本地調用，還是遠端調用是顯而易見的。
Document your system。清除元件之間的依賴關系。
處理錯誤情況。 當RPC伺服器長時間停機時，用戶端應該如何反應？
當有疑問時你應該避免使用RPC。 如果可以的話，你應該使用異步管道 - 而不是類似RPC的阻塞形式。采用異步管道的方式結果将被異步推送到下一個計算階段。
A note on RPC
雖然RPC是一個很常見的計算模式，但是它并不總是是一個很好的解決方式。當程式員不知道函數調用是本地完成還是通過緩慢的RPC時，可能會出現問題。 帶來未知的混亂，還對于調試增加了的不必要的複雜性。 而沒有達到不是簡化軟體的目的，是以濫用RPC可能導緻代碼不可維護。
銘記這一點，請考慮以下建議：
-
確定哪個函是在本地調用，還是遠端調用是顯而易見的。
Document your system。清除元件之間的依賴關系。
處理錯誤情況。 當RPC伺服器長時間停機時，用戶端應該如何反應？
2.Callback queue 回調隊列
In general doing RPC over RabbitMQ is easy. A client sends a request message and a server replies with a response message. In order to receive a response we need to send a 'callback' queue address with the request. We can use the default queue (which is exclusive in the Java client). Let's try it:
一般來說，RPC在RabbitMQ上實作很容易。 用戶端發送請求消息，伺服器回複響應消息。 為了用戶端能收到響應，我們發送請求時要包含一個‘callback’隊列的位址。 我們可以使用預設隊列（在Java用戶端中是排他性的）。 我們來試試吧
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 0-9-1協定預先定義了一組屬性（14個）。 大多數屬性很少使用，除了以下内容：
- deliveryMode: 将消息标記為持久性（值為2）或瞬态（任何其他值）。 您可能從 the second tutorial.中得知過此屬性。
- contentType: 用于描述MIME類型的編碼。 例如對于經常使用的JSON編碼，将此屬性設定為：application / json是一個很好的做法。
- replyTo: 通常用來命名一個 callback queue.
- correlationId: 用于将RPC響應與請求相關聯。
3.Correlation Id 關聯id
In the method presented above we suggest creating a callback queue for every RPC request. That's pretty inefficient, but fortunately there is a better way - let's create a single callback queue per client.
在上面提出的方法中，我們建議為每個RPC請求建立一個回調隊列。 這是非常低效的，但幸運的是有一個更好的方法 - 讓我們為每個用戶端建立一個回調隊列。
That raises a new issue, having received a response in that queue it's not clear to which request the response belongs. That's when the correlationId property is used. We're going to set it to a unique value for every request. Later, when we receive a message in the callback queue we'll look at this property, and based on that we'll be able to match a response with a request. If we see an unknown correlationId value, we may safely discard the message - it doesn't belong to our requests.
這引發了一個新的問題，callback隊列收到一個響應，但是不知道響應是屬于他自己的?
因為所有的響應都推送到了同一個callback隊列。上一小節的提到的correlation_id在這種情況下就派上用場了。 對于每個request，都設定唯一的辨別值。 之後，當我們在回調隊列中收到一條消息時，我們将檢視此屬性的值，這樣我們将能夠将響應與請求相比對。 如果我們看到一個未知的correlationId值，我們可能會安全地丢棄該消息 - 因為它不屬于我們的請求。
You may ask, why should we ignore unknown messages in the callback queue, rather than failing with an error? It's due to a possibility of a race condition on the server side. Although unlikely, it is possible that the RPC server will die just after sending us the answer, but before sending an acknowledgment message for the request. If that happens, the restarted RPC server will process the request again. That's why on the client we must handle the duplicate responses gracefully, and the RPC should ideally be idempotent.
您可能會問，為什麼我們應該忽略回調隊列中的未知消息，而不是抛出錯誤？ 這可能是由于伺服器端出現了問題。 RPC伺服器可能會在發送給我們的響應後，還沒有接收到發送響應請求的确認消息之前停止工作了。 如果發生這種情況，重新啟動的RPC伺服器将再次處理該請求。 這就是為什麼在用戶端上，我們必須優雅地處理這些重複的響應，而且RPC理應上是幂等的。
4.Summary 總結
-
- 當用戶端啟動時，它建立一個匿名的 anonymous 獨占的exclusive的callback隊列。
- 對于每個RPC請求，用戶端發送一個具有兩個屬性的消息：
- 請求然後被發送到rpc_queue隊列。
- RPC worker（aka：server）等待隊列上的請求。 一旦請求出現時，它将執行該作業，并使用replyTo字段中的辨別的隊列将結果發送回用戶端。
- 用戶端等待回調隊列中的資料。 當響應消息出現時，它檢查消息中的correlationId屬性。 如果它與某個請求中的設定的值相比對，則傳回對應用程式的響應。
5.實作Putting it all together最終實作
我們先定義fibonacci函數。 用來模拟耗時任務
The Fibonacci task:
private static int fib(int n) {
if (n == ) return ;
if (n == ) return ;
return fib(n-) + fib(n-);
}
We declare our fibonacci function. It assumes only valid positive integer input. (Don't expect this one to work for big numbers, and it's probably the slowest recursive implementation possible).
然後完成RPC服務端代碼
The code for our RPC server RPCServer.java looks like this:
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";
public static void main(String[] argv) {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("localhost");
Connection connection = null;
try {
connection = factory.newConnection();
Channel channel = connection.createChannel();
channel.queueDeclare(RPC_QUEUE_NAME, false, false, false, null);
channel.basicQos();
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"));
channel.basicAck(envelope.getDeliveryTag(), false);
}
}
};
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
//...
}
}
}
The server code is rather straightforward:
- As usual we start by establishing the connection, channel and declaring the queue.
- We might want to run more than one server process. In order to spread the load equally over multiple servers we need to set the prefetchCount setting in channel.basicQos.
- We use basicConsume to access the queue, where we provide a callback in the form of an object (DefaultConsumer) that will do the work and send the response back.
The code for our RPC client RPCClient.java:
import com.rabbitmq.client.*;
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();
}
public String call(String message) throws IOException, InterruptedException {
String corrId = UUID.randomUUID().toString();
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId)
.replyTo(replyQueueName)
.build();
channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));
final BlockingQueue<String> response = new ArrayBlockingQueue<String>();
channel.basicConsume(replyQueueName, true, new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
if (properties.getCorrelationId().equals(corrId)) {
response.offer(new String(body, "UTF-8"));
}
}
});
return response.take();
}
public void close() throws IOException {
connection.close();
}
//...
}
The client code is slightly more involved:
- We establish a connection and channel and declare an exclusive 'callback' queue for replies.
- We subscribe to the 'callback' queue, so that we can receive RPC responses.
- Our call method makes the actual RPC request.
- Here, we first generate a unique correlationId number and save it - our implementation of handleDelivery in DefaultConsumer will use this value to catch the appropriate response.
- Next, we publish the request message, with two properties: replyTo and correlationId.
- At this point we can sit back and wait until the proper response arrives.
- Since our consumer delivery handling is happening in a separate thread, we're going to need something to suspend main thread before response arrives. Usage of BlockingQueue is one of possible solutions. Here we are creating ArrayBlockingQueue with capacity set to 1 as we need to wait for only one response.
- The handleDelivery method is doing a very simple job, for every consumed response message it checks if the correlationId is the one we're looking for. If so, it puts the response to BlockingQueue.
- At the same time main thread is waiting for response to take it from BlockingQueue.
- Finally we return the response back to the user.
Making the Client request:
RPCClient fibonacciRpc = new RPCClient();
System.out.println(" [x] Requesting fib(30)");
String response = fibonacciRpc.call("30");
System.out.println(" [.] Got '" + response + "'");
fibonacciRpc.close();
Now is a good time to take a look at our full example source code (which includes basic exception handling) for RPCClient.java and RPCServer.java.
Compile and set up the classpath as usual (see tutorial one):
javac -cp $CP RPCClient.java RPCServer.java
Our RPC service is now ready. We can start the server:
java -cp $CP RPCServer
# => [x] Awaiting RPC requests
To request a fibonacci number run the client:
java -cp $CP RPCClient
# => [x] Requesting fib(30)
The design presented here is not the only possible implementation of a RPC service, but it has some important advantages:
- If the RPC server is too slow, you can scale up by just running another one. Try running a second RPCServer in a new console.
- On the client side, the RPC requires sending and receiving only one message. No synchronous calls like queueDeclare are required. As a result the RPC client needs only one network round trip for a single RPC request.
Our code is still pretty simplistic and doesn't try to solve more complex (but important) problems, like:
- How should the client react if there are no servers running?
- Should a client have some kind of timeout for the RPC?
- If the server malfunctions and raises an exception, should it be forwarded to the client?
- Protecting against invalid incoming messages (eg checking bounds, type) before processing.
If you want to experiment, you may find the management UI useful for viewing the queues.
![RabbitMQ：交換機（topic exchange）[圖]](data:image/gif;base64,R0lGODlhAQABAIAAAP///wAAACwAAAAAAQABAAACAkQBADs=)