1. 概述
本文使用RabbitMQ实现RPC的调用方式,主要包括如下内容:
- 回调队列(Callback queue)
- RPC调用相关的消息参数:replyTo和correlationId
- RPC调用的客户端和服务端的demo代码
2. 本文实现功能说明
本文使用RabbitMQ实现RPC的调用方式,我们需要使用新的队列:回调队列(Callback queue)
RPC需要涉及消息的两个重要属性:
- replyTo: 存储回调队列的名称
- correlationId: 唯一标识本次的请求,主要用于RPC调用
我们实现如上图的功能:
1. RPC客户端启动后,创建一个匿名、独占的、回调的队列
2. RPC客户端设置消息的2个属性:replyTo和correlationId,然后将消息发送到队列rpc_queue
3. RPC服务端在队列rpc_queue上等待消息。RPC服务端处理完收到消息后,然后将处理结果封装成消息发送到replyTo指定的队列上,并且此消息带上correlationId(此值为收到消息里的correlationId)
4. RPC客户端在队列replyTo上等待消息,当收到消息后,它会判断收到消息的correlationId。如果值和自己之前发送的一样,则这个值就是RPC的处理结果
3. RPC客户端: RpcClient代码
主要业务逻辑如下:
1. 配置连接工厂
2. 建立TCP连接
3. 在TCP连接的基础上创建通道
4. 定义临时队列replyQueueName,声明唯一标志本次请求corrId,并将replyQueueName和corrId配置要发送的消息队列中
5. 使用默认的交换机发送消息到队列rpc_queue中
6. 使用阻塞队列BlockingQueue阻塞当前进程
7. 收到请求后,将请求放入BlockingQueue中,主线程被唤醒,打印返回内容
下面只列出不同的地方: 第4,5步代码如下
// 定义临时队列,并返回生成的队列名称
String replyQueueName = channel.queueDeclare().getQueue();
// 唯一标志本次请求
String corrId = UUID.randomUUID().toString();
// 生成发送消息的属性
AMQP.BasicProperties props = new AMQP.BasicProperties
.Builder()
.correlationId(corrId) // 唯一标志本次请求
.replyTo(replyQueueName) // 设置回调队列
.build();
// 发送消息,发送到默认交换机
channel.basicPublish("", RPC_QUEUE_NAME, props, message.getBytes("UTF-8"));
第6,7步代码如下:
// 阻塞队列,用于存储回调结果
final BlockingQueue<String> response = new ArrayBlockingQueue<String>(1);
// 定义消息的回退方法
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"));
}
}
});
// 获取回调的结果
String result = response.take();
System.out.println(" [RpcClient] Result:'" + result + "'");
完整代码
RPC客户端代码:RpcClient.java
4. RPC服务端:RpcServer
主要业务逻辑如下:
1. 配置连接工厂
2. 建立TCP连接
3. 在TCP连接的基础上创建通道
4. 声明一个rpc_queue队列
5. 设置同时最多只能获取一个消息
6. 在rpc_queue队列在等待消息
7. 收到消息后,调用回调对象对消息进行处理,向此消息的replyTo队列中发送处理并带上correlationId
8. 使用wait-notify实现主线程和消息处理回调对象进行同步
下面只列出不同的地方: 第7,8步代码如下:
// 定义消息的回调处理类
Consumer consumer = new DefaultConsumer(channel) {
@Override
public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
// 生成返回的结果,关键是设置correlationId值
AMQP.BasicProperties replyProps = new AMQP.BasicProperties
.Builder()
.correlationId(properties.getCorrelationId())
.build();
// 生成返回
String response = generateResponse(body);
// 回复消息,通知已经收到请求
channel.basicPublish( "", properties.getReplyTo(), replyProps, response.getBytes("UTF-8"));
// 对消息进行应答
channel.basicAck(envelope.getDeliveryTag(), false);
// 唤醒正在消费者所有的线程
synchronized(this) {
this.notify();
}
}
};
// 消费消息
channel.basicConsume(RPC_QUEUE_NAME, false, consumer);
// 在收到消息前,本线程进入等待状态
while (true) {
synchronized(consumer) {
try {
consumer.wait();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
完整代码
RPC服务端代码: RpcServer.java
5. 测试:
BasicTest
@Test
public void rpc() throws InterruptedException {
// rpc服务端
executorService.submit(() -> {
RpcServer.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd);
});
// rpc客户端
executorService.submit(() -> {
RpcClient.execute(rabbitmq_host, rabbitmq_user, rabbitmq_pwd, "rpc test");
});
// sleep 10s
Thread.sleep(10 * 1000);
}
rpc客户端调用rpc服务端,并打印返回的结果
[RpcServer] Awaiting RPC requests
[RpcClient] Requesting : rpc test
[RpcServer] receive requests: rpc test
[RpcClient] Result:'response:rpc test-1516171567975
6. 代码
上文的详细代码主要如下:
RPC服务端代码: RpcServer.java
RPC客户端代码:RpcClient.java
测试代码:BasicTest.java的方法 header()
所有的详细代码见github代码,请尽量使用tag v0.12,不要使用master,因为master一直在变,不能保证文章中代码和github上的代码一直相同