1、RPC简述
RPC，Remote Procedure Call 远程过程调用。通俗讲，两段程序不在同一个内存空间，无法直接通过方法名调用，就需要通过网络通信方式调用。对于RabbitMQ，本身就是用于消息通信。简单的RabbitMQ是，生产端发送消息，经由交换器，到达队列。消费端不需要知道生产端，消费端订阅队列，消费队列中的消息。而对于RPC，希望消费端消费消息后，返回一个结果，结果经由网络，再返回给生产端。
不考虑RabbitMQ针对RPC的特有设计。最简单的设计是，生产端和消费端共同约定消费队列和回复队列，同时生产端每次发送消息时指定一个唯一ID。生产端将消息和唯一ID发送给消费队列，消费者从消费队列获取消息。处理后，将结果和生产端发送过来的唯一ID，发送给回复队列。生产端从回复队列获取消息和ID，判断ID是否匹配，匹配，则此消息为回复消息。
以上实现的RPC存在问题：生产端和消费端需要约定回复队列，这就要求生产端和消费端互相知道，这无法实现解耦。解决方案：生产端在发送消息时，也将回复队列名称随消息一起发送给队列。
2、举例说明RabbitMQ中的RPC实现
相关要点，见源码注释
pom依赖
<dependencies>
<dependency>
<groupId>com.rabbitmq</groupId>
<artifactId>amqp-client</artifactId>
<version>4.6.0</version>
</dependency>
</dependencies>
生产端，也就是客户端代码
package test; import java.io.IOException;
import java.util.UUID; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope; import utils.ChannelUtils; public class RPCClient { public static void main(String[] args) throws IOException {
RPCClient rpcClient = new RPCClient();
rpcClient.client();
} public void client() throws IOException {
//此方法封装了如何连接RabbitMQ和创建connection，channel.源码见附录
Channel channel = ChannelUtils.getChannelInstance("client");
channel.exchangeDelete("exchange_rpc");
channel.exchangeDeclare("exchange_rpc", "direct", false, false, null); channel.queueDelete("queue_rpc");
channel.queueDeclare("queue_rpc", false, false, false, null); channel.queueBind("queue_rpc", "exchange_rpc", "rpc"); //此处注意：我们声明了要回复的队列。队列名称由RabbitMQ自动创建。
//这样做的好处是：每个客户端有属于自己的唯一回复队列，生命周期同客户端
String replyQueue = channel.queueDeclare().getQueue();
final String corrID = UUID.randomUUID().toString(); //这里我们设计三类消息。
//消息1：指定回复队列和ID
//消息2：仅指定回复队列
//消息3：不指定回复队列和ID
AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
// 指定回复队列和回复correlateId
builder.replyTo(replyQueue).correlationId(corrID);
AMQP.BasicProperties properties = builder.build();
for (int i = 0; i < 2; i++) {
channel.basicPublish("exchange_rpc", "rpc", properties,
(System.currentTimeMillis() + "-rpc发送消息1").getBytes());
} AMQP.BasicProperties.Builder builder2 = new AMQP.BasicProperties.Builder();
// 指定回复队列，未指定回复correlateId
builder2.replyTo(replyQueue);
AMQP.BasicProperties properties2 = builder2.build();
for (int i = 0; i < 2; i++) {
channel.basicPublish("exchange_rpc", "rpc", properties2,
(System.currentTimeMillis() + "-rpc发送消息2").getBytes());
} for (int i = 0; i < 2; i++) {
// 未指定回复队列和correlateId
channel.basicPublish("exchange_rpc", "rpc", null, (System.currentTimeMillis() + "-rpc发送消息3").getBytes());
} DefaultConsumer c = new DefaultConsumer(channel) {
//这是一个回调函数，客户端获取消息，就调用此方法，处理消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
if (corrID.equals(properties.getCorrelationId())) {
System.out.println("correlationID对应上的消息：" + new String(body));
} else {
System.out.println("correlationID未对应上的消息：" + new String(body));
}
}
};
channel.basicConsume(replyQueue, true, c);
} }
消费端，也就是服务器端代码
package test; import java.io.IOException; import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.AMQP.BasicProperties; import utils.ChannelUtils; public class RPCServer { public static void main(String[] args) throws IOException {
RPCServer rpcServer = new RPCServer();
rpcServer.Server();
} public void Server() throws IOException {
final Channel channel = ChannelUtils.getChannelInstance("server"); DefaultConsumer c = new DefaultConsumer(channel) { //这是一个回到函数，服务器端获取到消息，就会调用此方法处理消息
@Override
public void handleDelivery(String consumerTag, Envelope envelope, BasicProperties properties, byte[] body)
throws IOException {
System.out.println(new String(body)); AMQP.BasicProperties.Builder builder = new AMQP.BasicProperties.Builder();
//我们在将要回复的消息属性中，放入从客户端传递过来的correlateId
builder.correlationId(properties.getCorrelationId());
AMQP.BasicProperties prop = builder.build(); //发送给回复队列的消息，exchange=""，routingKey=回复队列名称
//因为RabbitMQ对于队列，始终存在一个默认exchange=""，routingKey=队列名称的绑定关系
channel.basicPublish("", properties.getReplyTo(), prop, (new String(body) + "-回复").getBytes()); }
};
channel.basicConsume("queue_rpc", true, c);
} }
附录：ChannelUtils 的源码
package utils;import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;public class ChannelUtils { // AMQP的连接其实是对Socket做的封装, 注意以下AMQP协议的版本号，不同版本的协议用法可能不同。
public static Channel getChannelInstance(String ConnectionDescription) {
try {
ConnectionFactory connectionFactory = getConnectionFactory();
Connection connection = connectionFactory.newConnection(ConnectionDescription); return connection.createChannel();
} catch (Exception e) {
throw new RuntimeException("获取Channel连接失败");
} } public static ConnectionFactory getConnectionFactory() {
ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost("192.168.1.111");
connectionFactory.setPort(5672);
connectionFactory.setVirtualHost("/");
connectionFactory.setUsername("drs");
connectionFactory.setPassword("123456"); return connectionFactory;
}}
