RabbitMQ工作模式中的RPC通信模式详解

2025-06-03 16:50

本文主要是介绍RabbitMQ工作模式中的RPC通信模式详解,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

《RabbitMQ工作模式中的RPC通信模式详解》在RabbitMQ中,RPC模式通过消息队列实现远程调用功能,这篇文章给大家介绍RabbitMQ工作模式之RPC通信模式,感兴趣的朋友一起看看吧...

RPC通信模式

概述

在RabbitMQ中,RPC模式通过消息队列实现远程调用功能。客户端(生产者)发送消息到消费队列,服务端(消费者)进行消息消费并执行相应的程序,然后将结果发送到回调队列供客户端使用。这是一种双向的生产消费模式,其中客户端既是生产者又是消费者,服务端则专注于处理消息并生成响应。

RabbitMQ工作模式中的RPC通信模式详解

在RPC通信的过程中, 没有⽣产者和消费者, ⽐较像咱们RPC远程调⽤, ⼤概就是通过两个队列实现了⼀个可回调的过程.

工作流程

RabbitMQ工作模式中的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-cliChina编程ent</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.setPasswojavascriptrd(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. 接收响应
        //使用阻塞队列, 来存储响应信息
      js  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_REphpQUEST_QUEUE, false, consumer);
    }
}

运行程序(先运行客户端,再运行服务端)

可以在管理界面看到其中一个队列中有1条消息

RabbitMQ工作模式中的RPC通信模式详解

RabbitMQ工作模式中的RPC通信模式详解

RabbitMQ工作模式中的RPC通信模式详解

我们可以看到,服务端接收到了消息并给客户端发送了响应,与预期符合。

到此这篇关于RabbitMQ工作模式之RPC通信模式的文章就介绍到这了,更多相关RabbitMQ RPC通信模式内容请搜索China编程(www.chinasem.cn)以前的文章或继续浏览下面的相关文章希望大家以后多多支持编程China编程(www.chinasem.cn)!

这篇关于RabbitMQ工作模式中的RPC通信模式详解的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



http://www.chinasem.cn/article/1154889

相关文章

Windows 系统下 Nginx 的配置步骤详解

《Windows系统下Nginx的配置步骤详解》Nginx是一款功能强大的软件,在互联网领域有广泛应用,简单来说,它就像一个聪明的交通指挥员,能让网站运行得更高效、更稳定,:本文主要介绍W... 目录一、为什么要用 Nginx二、Windows 系统下 Nginx 的配置步骤1. 下载 Nginx2. 解压

详解如何使用Python从零开始构建文本统计模型

《详解如何使用Python从零开始构建文本统计模型》在自然语言处理领域,词汇表构建是文本预处理的关键环节,本文通过Python代码实践,演示如何从原始文本中提取多尺度特征,并通过动态调整机制构建更精确... 目录一、项目背景与核心思想二、核心代码解析1. 数据加载与预处理2. 多尺度字符统计3. 统计结果可

Linux基础命令@grep、wc、管道符的使用详解

《Linux基础命令@grep、wc、管道符的使用详解》:本文主要介绍Linux基础命令@grep、wc、管道符的使用,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐... 目录grep概念语法作用演示一演示二演示三,带选项 -nwc概念语法作用wc,不带选项-c,统计字节数-

SpringCloud中的@FeignClient注解使用详解

《SpringCloud中的@FeignClient注解使用详解》在SpringCloud中使用Feign进行服务间的调用时,通常会使用@FeignClient注解来标记Feign客户端接口,这篇文章... 在Spring Cloud中使用Feign进行服务间的调用时,通常会使用@FeignClient注解

Java Spring 中的监听器Listener详解与实战教程

《JavaSpring中的监听器Listener详解与实战教程》Spring提供了多种监听器机制,可以用于监听应用生命周期、会话生命周期和请求处理过程中的事件,:本文主要介绍JavaSprin... 目录一、监听器的作用1.1 应用生命周期管理1.2 会话管理1.3 请求处理监控二、创建监听器2.1 Ser

maven中的maven-antrun-plugin插件示例详解

《maven中的maven-antrun-plugin插件示例详解》maven-antrun-plugin是Maven生态中一个强大的工具,尤其适合需要复用Ant脚本或实现复杂构建逻辑的场景... 目录1. 核心功能2. 典型使用场景3. 配置示例4. 关键配置项5. 优缺点分析6. 最佳实践7. 常见问题

JVisualVM之Java性能监控与调优利器详解

《JVisualVM之Java性能监控与调优利器详解》本文将详细介绍JVisualVM的使用方法,并结合实际案例展示如何利用它进行性能调优,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全... 目录1. JVisualVM简介2. JVisualVM的安装与启动2.1 启动JVisualVM2

Redis中的Lettuce使用详解

《Redis中的Lettuce使用详解》Lettuce是一个高级的、线程安全的Redis客户端,用于与Redis数据库交互,Lettuce是一个功能强大、使用方便的Redis客户端,适用于各种规模的J... 目录简介特点连接池连接池特点连接池管理连接池优势连接池配置参数监控常用监控工具通过JMX监控通过Pr

在Spring Boot中实现HTTPS加密通信及常见问题排查

《在SpringBoot中实现HTTPS加密通信及常见问题排查》HTTPS是HTTP的安全版本,通过SSL/TLS协议为通讯提供加密、身份验证和数据完整性保护,下面通过本文给大家介绍在SpringB... 目录一、HTTPS核心原理1.加密流程概述2.加密技术组合二、证书体系详解1、证书类型对比2. 证书获

MySQL 添加索引5种方式示例详解(实用sql代码)

《MySQL添加索引5种方式示例详解(实用sql代码)》在MySQL数据库中添加索引可以帮助提高查询性能,尤其是在数据量大的表中,下面给大家分享MySQL添加索引5种方式示例详解(实用sql代码),... 在mysql数据库中添加索引可以帮助提高查询性能,尤其是在数据量大的表中。索引可以在创建表时定义,也可