Rabbitmq远程过程调用（RPC）模式
目录
前面的几种模式都是生产者发布了消息到达队列之后，这些消息如何被处理就不管了，现在RPC模式就是客户端发送RPC请求并进入阻塞状态，直到收到服务端处理这些消息之后并返回答案为止，在远程计算机上运行功能并等待结果这种模式通常称为“ 远程过程调用”或“ RPC”。
回调队列：客户端发送请求消息到服务器使用了一个队列，然而服务器要向客户端发送响应消息也需要一个队列。所以我们需要客户端要使用replyTo方法设置一个临时队列给服务端使用。
relatedId（关联ID）：将为每个请求将其设置为唯一值,当我们在客户端收到响应的消息时，响应的relatedId与请求的relatedId进行比较。如果不相等值，我们可以放心地丢弃该消息-它不属于我们的请求。
例如下面代码中的if (delivery.getProperties().getCorrelationId().equals(corrId))
开始演示：
第一步、执行服务端RPCServer.java，监听客户端RPC请求。
第二步、执行客户端RPCClient.java，发送5个RPC请求，并阻塞等待服务端响应返回数据。
客户端打印的结果：
客户端开始发送消息：0
远程服务端响应的内容：本服务已处理的消息【0】
---------------------------------------------------------------------------
客户端开始发送消息：1
远程服务端响应的内容：本服务已处理的消息【1】
---------------------------------------------------------------------------
客户端开始发送消息：2
远程服务端响应的内容：本服务已处理的消息【2】
---------------------------------------------------------------------------
客户端开始发送消息：3
远程服务端响应的内容：本服务已处理的消息【3】
---------------------------------------------------------------------------
客户端开始发送消息：4
远程服务端响应的内容：本服务已处理的消息【4】
服务端打印的结果：
服务器启动成功，正等待客户端RPC请求......
服务端收到的RPC请求：0
服务端收到的RPC请求：1
服务端收到的RPC请求：2
服务端收到的RPC请求：3
服务端收到的RPC请求：4
演示结论：
客户端发送RPC请求并进入阻塞状态，直到收到服务端处理这些消息之后并返回答案为止。
客户端代码：
package com.demo;
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("127.0.0.1");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
connection = factory.newConnection();
channel = connection.createChannel();
}
public static void main(String[] argv) {
try (RPCClient fibonacciRpc = new RPCClient()) {
for (int i = 0; i < 5; i++) {
String i_str = Integer.toString(i);
System.out.println("客户端开始发送消息：" + i_str );
String response = fibonacciRpc.call(i_str);
System.out.println("远程服务端响应的内容：" + response);
System.out.println("---------------------------------------------------------------------------");
}
} 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<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();
}
}
服务端代码：
package com.demo;
import com.rabbitmq.client.*;
public class RPCServer {
private static final String RPC_QUEUE_NAME = "rpc_queue";
public static void main(String[] argv) throws Exception {
ConnectionFactory factory = new ConnectionFactory();
factory.setHost("127.0.0.1");
factory.setPort(5672);
factory.setUsername("guest");
factory.setPassword("guest");
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("服务器启动成功，正等待客户端RPC请求......");
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");
System.out.println("服务端收到的RPC请求：" + message);
response += "本服务已处理的消息【"+message+"】";
} catch (RuntimeException e) {
e.printStackTrace();
} finally {
channel.basicPublish("", delivery.getProperties().getReplyTo(), replyProps, response.getBytes("UTF-8"));
channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
synchronized (monitor) {
monitor.notify();
}
}
};
channel.basicConsume(RPC_QUEUE_NAME, false, deliverCallback, (consumerTag -> { }));
while (true) {
synchronized (monitor) {
try {
monitor.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
}
}