详解RabbitMQ工作模式之RPC通信模式

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){@Overridepublic 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){@Overridepublic 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);}
}