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

- 日理万妓 2022-06-01 12:36 163阅读 0赞

## 1. 概述 ##

本文使用RabbitMQ实现RPC的调用方式，主要包括如下内容：

*  回调队列(Callback queue)
 *  RPC调用相关的消息参数：replyTo和correlationId
 *  RPC调用的客户端和服务端的demo代码

## 2. 本文实现功能说明 ##

本文使用RabbitMQ实现RPC的调用方式，我们需要使用新的队列：回调队列(Callback queue)

RPC需要涉及消息的两个重要属性：

*  replyTo: 存储回调队列的名称
 *  correlationId: 唯一标识本次的请求，主要用于RPC调用

![这里写图片描述][SouthEast]

我们实现如上图的功能：

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上的代码一直相同][github_tag v0.12_master_master_github]

[SouthEast]: /images/20220601/c42ea22d41374f4598f398cbfda16e16.png
[RpcClient.java]: https://github.com/hryou0922/spring_boot/blob/v0.12/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/rpc/RpcClient.java
[RpcServer.java]: https://github.com/hryou0922/spring_boot/blob/v0.12/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/rpc/RpcServer.java
[BasicTest.java]: https://github.com/hryou0922/spring_boot/blob/v0.12/rabbitmq/src/test/java/com/hry/spring/rabbitmq/basic/BasicTest.java
[github_tag v0.12_master_master_github]: https://github.com/hryou0922/spring_boot/tree/v0.12/rabbitmq/src/main/java/com/hry/spring/rabbitmq/basic/rpc