专栏名称: hryou0922

中间件系列八 RabbitMQ之实现RPC模式

hryou0922 · 掘金 · · 2018-02-05 02:42

正文

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