SpringBoot整合RabbitMQ实现RPC远程调用功能

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>2.3.10.RELEASE</version>
    <relativePath/>
</parent>
<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
        <version>2.3.10.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-amqp</artifactId>
        <version>2.3.10.RELEASE</version>
    </dependency>
    <dependency>
        <groupId>com.rabbitmq</groupId>
        <artifactId>amqp-client</artifactId>
        <version>5.12.0</version>
    </dependency>
</dependencies>

spring:
  rabbitmq:
    host: localhost # 主机名
    port: 5672 # 端口
    username: guest # 用户名
    password: guest # 密码
    virtual-host: /
    template:
      receive-timeout: 2000
      reply-timeout: 2000
    listener:
      simple:
        concurrency: 1
        max-concurrency: 3
        prefetch: 1 # 消费者每次只能预取1条数据到内存并处理，默认为250条
        acknowledge-mode: manual # 确定机制 manual：手动确认
    publisher-returns: true
    publisher-confirm-type: correlated

package com.rabbit.rpc;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeoutException;
/**
 * rpc服务端
 */
public class RPCServer {
    // 定义请求队列常量
    private final static String REQUEST_QUEUE_NAME = "rpc_queue";
    // 定义回复队列常量
    private final static String REPLY_QUEUE_NAME = "rpc_reply_queue";
    /**
     * 服务端启动入口
     */
    public static void main(String[] args) {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        //factory.setHost("localhost");
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明请求队列和回复队列
            channel.queueDeclare(REQUEST_QUEUE_NAME, false, false, false, null);
            channel.queueDeclare(REPLY_QUEUE_NAME, false, false, false, null);
            // 每次仅接收一条未经确认的消息
            channel.basicQos(1);
            // 构建消费者属性
            Consumer consumer = new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // 获取消息
                    String message = new String(body, StandardCharsets.UTF_8);
                    // 进行业务处理构建响应数据 (这里做字符串拼接模拟响应数据)
                    String response = message + ":::";
                    // 构造响应基本属性
                    AMQP.BasicProperties replyProps = new AMQP.BasicProperties.Builder()
                            .correlationId(properties.getCorrelationId())   // 设置关联id
                            .build();
                    // 发送响应数据到回复队列 (rpc_reply_queue)
                    channel.basicPublish("", properties.getReplyTo(), replyProps, response.getBytes(StandardCharsets.UTF_8));
                    // 手动回执消息确认消费
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            };
            // 消费rpc_queue队列的消息
            channel.basicConsume(REQUEST_QUEUE_NAME, false, consumer);
            // 持续监听请求消息
            while (true) {
                Thread.sleep(50);
            }
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}

package com.rabbit.rpc;
import com.rabbitmq.client.*;
import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeoutException;
/**
 * rpc客户端
 */
public class RPCClient {
    // 定义请求队列常量
    private final static String REQUEST_QUEUE_NAME = "rpc_queue";
    // 定义回复队列常量
    private final static String REPLY_QUEUE_NAME = "rpc_reply_queue";
    /**
     * 客户端启动入口
     */
    public static void main(String[] args) {
        System.out.println("Response1: " + call("Hello, RPC Server1"));
        System.out.println("Response2: " + call("Hello, RPC Server2"));
        System.out.println("Response3: " + call("Hello, RPC Server3"));
    }
    /**
     * 发送请求到队列并返回响应数据
     *
     * @param message 请求消息
     * @return 响应数据
     */
    public static String call(String message) {
        // 创建连接
        ConnectionFactory factory = new ConnectionFactory();
        //factory.setHost("localhost");
        String result = "";
        try (Connection connection = factory.newConnection();
             Channel channel = connection.createChannel()) {
            // 声明请求队列和回复队列
            channel.queueDeclare(REQUEST_QUEUE_NAME, false, false, false, null);
            channel.queueDeclare(REPLY_QUEUE_NAME, false, false, false, null);
            // 每次仅接收一条未经确认的消息
            channel.basicQos(1);
            // 生成关联id
            String correlationId = UUID.randomUUID().toString();
            // 构造请求基本属性
            AMQP.BasicProperties props = new AMQP.BasicProperties.Builder()
                    .correlationId(correlationId)   // 设置关联id
                    .replyTo(REPLY_QUEUE_NAME)      // 设置回复队列 (rpc_reply_queue)
                    .build();
            // 发送请求消息到rpc_queue队列
            channel.basicPublish("", REQUEST_QUEUE_NAME, props, message.getBytes(StandardCharsets.UTF_8));
            // 用于保存响应消息的阻塞队列
            final BlockingQueue<String> response = new ArrayBlockingQueue<>(1);
            // 监听回复队列接收响应消息
            String consumerTag = channel.basicConsume(REPLY_QUEUE_NAME, true, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) {
                    // 通过关联id获取到响应数据
                    if (properties.getCorrelationId().equals(correlationId)) {
                        response.offer(new String(body, StandardCharsets.UTF_8));
                    }
                }
            });
            // 清空阻塞队列
            response.clear();
            // 等待接收响应消息
            result = response.take();
            // 取消消费者的监听
            channel.basicCancel(consumerTag);
        } catch (IOException | TimeoutException | InterruptedException e) {
            e.printStackTrace();
        }
        return result;
    }
}