详解RabbitMQ工作模式之RPC通信模式
目录
RPC通信模式
概述
在RabbitMQ中，RPC模式通过消息队列实现远程调用功能。客户端（生产者）发送消息到消费队列，服务端（消费者）进行消息消费并执行相应的程序，然后将结果发送到回调队列供客户端使用。这是一种双向的生产消费模式，其中客户端既是生产者又是消费者，服务端则专注于处理消息并生成响应。
在RPC通信的过程中, 没有⽣产者和消费者, ⽐较像咱们RPC远程调⽤, ⼤概就是通过两个队列实现了⼀个可回调的过程.
工作流程
1.客户端发送请求：
客户端连接到RabbitMQ服务器。
客户端声明一个用于发送RPC请求的队列（通常是固定的，如rpc_queue）。
客户端创建一个临时的回调队列，并在发送请求时，将回调队列的名称作为消息属性（reply_to）发送给交换机。
客户端为每个请求生成一个唯一的correlation_id，并将其作为消息属性发送，以便在接收响应时能够匹配请求与响应。
2.交换机路由请求：
交换机接收到RPC请求后，根据路由键将请求路由到服务端监听的队列。
3.服务端处理请求：
服务端（消费者）从队列中接收请求。
服务端处理请求，并生成响应。
服务端将响应发送到客户端指定的回调队列，并在消息属性中设置相同的correlation_id。
4.客户端接收响应：
客户端监听其回调队列以接收响应。
当接收到响应时，客户端检查correlation_id以确定响应是否与之前的请求匹配。
如果匹配，客户端处理响应；如果不匹配，客户端可能丢弃该响应。
特点
1.解耦：客户端和服务端之间不需要直接通信，降低了系统间的耦合度。
2.灵活性：支持多种语言和平台之间的远程调用。
3.可扩展性：通过增加服务端（消费者）的数量，可以轻松扩展RPC服务。
4.性能开销：由于涉及到网络传输和消息队列的处理，RPC调用的性能通常低于本地调用。
5.复杂性：需要处理消息队列的可靠性、持久性、消息确认等复杂问题。
6.安全性：远程调用可能面临更多的安全风险，如消息篡改、中间人攻击等。
应用场景
RabbitMQ的RPC通信模式适用于需要远程调用服务的场景，如分布式系统中的服务调用、微服务架构中的服务通信等。通过RabbitMQ的消息队列机制，可以实现跨系统、跨语言的远程调用，提高系统的灵活性和可扩展性。
代码案例
引入依赖
<!-- https://mvnrepository.com/artifact/com.rabbitmq/amqp-client --> <dependency> <groupId>com.rabbitmq</groupId> <artifactId>amqp-client</artifactId> <version>5.21.0</version> </dependency>
常量类
public class Constants { public static final String HOST = "47.98.109.138"; public static final int PORT = 5672; public static final String USER_NAME = "study"; public static final String PASSWORD = "study"; public static final String VIRTUAL_HOST = "aaa"; //rpc 模式 public static final String RPC_REQUEST_QUEUE = "rpc.request.queue"; public static final String RPC_RESPONSE_QUEUE = "rpc.response.queue"; }编写客户端代码
import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.UUID; import java.util.concurrent.ArrayBlockingQueue; import java.util.concurrent.BlockingQueue; import java.util.concurrent.TimeoutException; /** * rpc 客户端 * 1. 发送请求 * 2. 接收响应 */ public class RpcClient { public static void main(String[] args) throws IOException, TimeoutException, InterruptedException { //1. 建立连接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); //需要提前开放端口号 connectionFactory.setUsername(Constants.USER_NAME);//账号 connectionFactory.setPassword(Constants.PASSWORD); //密码 connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机 Connection connection = connectionFactory.newConnection(); //2. 开启信道 Channel channel = connection.createChannel(); channel.queueDeclare(Constants.RPC_REQUEST_QUEUE, true, false, false, null); channel.queueDeclare(Constants.RPC_RESPONSE_QUEUE, true, false, false, null); //3. 发送请求 String msg = "hello rpc..."; //设置请求的唯一标识 String correlationID = UUID.randomUUID().toString(); //设置请求的相关属性 AMQP.BasicProperties props = new AMQP.BasicProperties().builder() .correlationId(correlationID) .replyTo(Constants.RPC_RESPONSE_QUEUE) .build(); channel.basicPublish("", Constants.RPC_REQUEST_QUEUE, props, msg.getBytes()); //4. 接收响应 //使用阻塞队列, 来存储响应信息 final BlockingQueue<String> response = new ArrayBlockingQueue<>(1); DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String respMsg = new String(body); System.out.println("接收到回调消息: "+ respMsg); if (correlationID.equals(properties.getCorrelationId())){ //如果correlationID校验一致 response.offer(respMsg); } } }; channel.basicConsume(Constants.RPC_RESPONSE_QUEUE, true, consumer); String result = response.take(); System.out.println("[RPC Client 响应结果]:"+ result); } }编写服务端代码
import com.rabbitmq.client.*; import rabbitmq.constant.Constants; import java.io.IOException; import java.util.concurrent.TimeoutException; /** * RPC server * 1. 接收请求 * 2. 发送响应 */ public class RpcServer { public static void main(String[] args) throws IOException, TimeoutException { //1. 建立连接 ConnectionFactory connectionFactory = new ConnectionFactory(); connectionFactory.setHost(Constants.HOST); connectionFactory.setPort(Constants.PORT); //需要提前开放端口号 connectionFactory.setUsername(Constants.USER_NAME);//账号 connectionFactory.setPassword(Constants.PASSWORD); //密码 connectionFactory.setVirtualHost(Constants.VIRTUAL_HOST); //虚拟主机 Connection connection = connectionFactory.newConnection(); //2. 开启信道 Channel channel = connection.createChannel(); //3. 接收请求 channel.basicQos(1); DefaultConsumer consumer = new DefaultConsumer(channel){ @Override public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException { String request = new String(body,"UTF-8"); System.out.println("接收到请求:"+ request); String response = "针对request:"+ request +", 响应成功"; AMQP.BasicProperties basicProperties = new AMQP.BasicProperties().builder() .correlationId(properties.getCorrelationId()) .build(); channel.basicPublish("", Constants.RPC_RESPONSE_QUEUE, basicProperties, response.getBytes()); channel.basicAck(envelope.getDeliveryTag(), false); } }; channel.basicConsume(Constants.RPC_REQUEST_QUEUE, false, consumer); } }运行程序（先运行客户端，再运行服务端）
可以在管理界面看到其中一个队列中有1条消息
我们可以看到，服务端接收到了消息并给客户端发送了响应，与预期符合。