消息中间件RabbitMQ（五）——实现RPC调用

article/2025/10/4 2:09:47

文章目录

  • 1. RPC
  • 2. 实现原理
  • 3. 代码实现
    • 3.1 客户端实现
    • 3.2 服务端实现
    • 3.3 测试
  • 4. 小结

1. RPC

对于微服务开发者，对于 RPC（Remote Procedure Call Protocol 远程过程调用协议）并不会陌生吧， RESTful API、Dubbo、WebService等都是RPC的实现调用

在RabbitMQ中也提供了 RPC 功能，并且使用起来很简单，下面就来学习一下

2. 实现原理

再来熟悉下原理图

在这里插入图片描述

上图把RPC的过程描述的很清楚：

  • Client先发送一条消息，和普通的消息相比，消息多了两个关键内容：一个是 correlation_id，表示这条消息的唯一 id，一个是 reply_to，表示回复队列的名字
  • Server从消息发送队列获取消息并处理相应的业务逻辑，处理完成后，将处理结果发送到 reply_to指定的回调队列中
  • Client从回调队列中读取消息，就可知道执行结果

3. 代码实现

3.1 客户端实现

客户端配置文件：application.properties

server.port=8889
spring.rabbitmq.host=192.168.3.157
spring.rabbitmq.port=5672
spring.rabbitmq.username=guest
spring.rabbitmq.password=guest
spring.rabbitmq.virtual-host=/
# 开启消息确认
spring.rabbitmq.publisher-confirm-type=correlated
# 开启发送失败退回
spring.rabbitmq.publisher-returns=true

spring.rabbitmq.publisher-confirm-type=correlated这项配置作用是：通过 correlated来确认消息。

只有开启了这个配置，将来的消息中才会带 correlation_id，只有通过 correlation_id才能将发送的消息和返回值之间关联起来

客户端配置类：

package com.scorpios.rabbitmq.config;import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;@Configuration
public class RPCRabbitMQConfig {// 交换机的名称public static final String SCORPIOS_RPC_EXCHANGE_NAME = "scorpios_rpc_exchange_name";// 发送队列名称public static final String SCORPIOS_RPC_MSG_QUEUE = "scorpios_rpc_msg_queue";// 返回队列名称public static final String SCORPIOS_RPC_REPLY_QUEUE = "scorpios_rpc_reply_queue";@BeanTopicExchange topicExchange(){return new TopicExchange(RPCRabbitMQConfig.SCORPIOS_RPC_EXCHANGE_NAME,true,false);}@BeanQueue queueOne() {return new Queue(RPCRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE,true,false,false);}@BeanQueue queueTwo() {return new Queue(RPCRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE,true,false,false);}/*** 请求队列和交换器绑定*/@BeanBinding bindingMsg(){return BindingBuilder.bind(queueOne()).to(topicExchange()).with(RPCRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE);}/*** 返回队列和交换器绑定*/@BeanBinding bindingReply(){return BindingBuilder.bind(queueTwo()).to(topicExchange()).with(RPCRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE);}/*** 自定义 RabbitTemplate发送和接收消息，因为要设置回调队列地址*/@BeanRabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory){RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);rabbitTemplate.setReplyAddress(RPCRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE);rabbitTemplate.setReplyTimeout(5000);return rabbitTemplate;}/*** 给返回队列设置监听器*/@BeanSimpleMessageListenerContainer replyContainer(ConnectionFactory connectionFactory) {SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();container.setConnectionFactory(connectionFactory);container.setQueueNames(RPCRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE);container.setMessageListener(rabbitTemplate(connectionFactory));return container;}}

上面代码解释说明：

  • 定义一个TopicExchange交换机，一个MsgQueue队列，一个ReplyQueue，并与交换机进行绑定
  • 自定义一个RabbitTemplate用户发送消息，虽然在 SpringBoot中，默认情况下系统自动提供RabbitTemplate，但是这里需要对该RabbitTemplate重新进行定制，因为要给RabbitTemplate添加返回队列，最后还需要给返回队列设置一个监听器

下面来编写消息发送代码：

@Slf4j
@RestController
public class RabbitMQController {@Autowiredprivate RabbitTemplate rabbitTemplate;@GetMapping("/send/message")public String send(String message) {// 创建消息对象Message newMessage = MessageBuilder.withBody(message.getBytes()).build();log.info("Client 发送的消息为：{}", newMessage);// 客户端给消息队列发送消息，并返回响应结果Message result = rabbitTemplate.sendAndReceive(RPCRabbitMQConfig.SCORPIOS_RPC_EXCHANGE_NAME, RPCRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE, newMessage);String response = "";if (result != null) {// 获取已发送的消息的 correlationIdString correlationId = newMessage.getMessageProperties().getCorrelationId();log.info("发送消息的correlationId为：{}", correlationId);// 获取响应头信息HashMap<String, Object> headers = (HashMap<String, Object>) result.getMessageProperties().getHeaders();// 获取 server 返回的消息 correlationIdString msgId = (String) headers.get("spring_returned_message_correlation");// 将已发送的消息的 correlationId与server返回的消息 correlationId进行对比，相同则取出响应结果if (msgId.equals(correlationId)) {response = new String(result.getBody());log.info("client 收到的响应结果为：{}", response);}}return response;}}

解释说明：

  • 消息发送调用 sendAndReceive方法，该方法自带返回值，返回值就是服务端返回的消息
  • 服务端返回的消息中，头信息中包含了 spring_returned_message_correlation字段，这就是消息发送时的 correlation_id，通过消息发送时的 correlation_id以及返回消息头中的 spring_returned_message_correlation字段值，就可以将返回的消息内容和发送的消息绑定到一起，确认出这个返回的内容就是针对这个发送的消息的

注意：如果没有在 application.properties 中配置 correlated，发送的消息中就没有 correlation_id，这样就无法将返回的消息内容和发送的消息内容关联起来

3.2 服务端实现

服务端配置文件 application.properties与客户端中的配置文件一致

服务端配置类：

@Configuration
public class RPCServerRabbitMQConfig {// 交换机的名称public static final String SCORPIOS_RPC_EXCHANGE_NAME = "scorpios_rpc_exchange_name";// 发送队列名称public static final String SCORPIOS_RPC_MSG_QUEUE = "scorpios_rpc_msg_queue";// 返回队列名称public static final String SCORPIOS_RPC_REPLY_QUEUE = "scorpios_rpc_reply_queue";@BeanTopicExchange topicExchange(){return new TopicExchange(RPCServerRabbitMQConfig.SCORPIOS_RPC_EXCHANGE_NAME,true,false);}@BeanQueue queueOne() {return new Queue(RPCServerRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE,true,false,false);}@BeanQueue queueTwo() {return new Queue(RPCServerRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE,true,false,false);}@BeanBinding bindingMsg(){return BindingBuilder.bind(queueOne()).to(topicExchange()).with(RPCServerRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE);}@BeanBinding bindingReply(){return BindingBuilder.bind(queueTwo()).to(topicExchange()).with(RPCServerRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE);}}

最后我们再来看下消息的消费：

@Slf4j
@Component
public class RpcServerConsumer {@Autowiredprivate RabbitTemplate rabbitTemplate;// 此消费者消费msgQueue队列中的消息@RabbitListener(queues = RPCServerRabbitMQConfig.SCORPIOS_RPC_MSG_QUEUE)public void process(Message msg) {log.info("server 收到msgQueue队列中的消息为 : {}",msg.toString());Message response = MessageBuilder.withBody(("我是服务端Server，收到的消息为:"+new String(msg.getBody())).getBytes()).build();// 把收到的原消息的CorrelationId取出CorrelationData correlationData = new CorrelationData(msg.getMessageProperties().getCorrelationId());// 想replyQueue队列发送确认消息rabbitTemplate.sendAndReceive(RPCServerRabbitMQConfig.SCORPIOS_RPC_EXCHANGE_NAME, RPCServerRabbitMQConfig.SCORPIOS_RPC_REPLY_QUEUE, response, correlationData);}}

解释说明：

  • 服务端首先收到消息并打印出来
  • 服务端提取出原消息中的 correlation_id
  • 服务端调用 sendAndReceive方法，将消息发送给 replyQueue队列，同时带上 correlation_id参数

3.3 测试

启动Client与Server服务，并在浏览器中输入：http://localhost:8889/send/scorpios

Client服务日志：

在这里插入图片描述

Server服务日志：

在这里插入图片描述

浏览器响应结果：
在这里插入图片描述

4. 小结

再来看一下这个原理图：

在这里插入图片描述

  • 定义一个Exchange交换机，两个队列：MsgQueue、ReplyQueue
  • Client调用 sendAndReceive方法向MsgQueue队列中发送消息，该方法自带返回值，返回值就是服务端返回的消息
  • 在Server端消费MsgQueue队列消息后，往ReplayQueue中发送消息

代码地址：https://github.com/Hofanking/springboot-rabbitmq-example

springboot-rabbitmq-rpc-client

springboot-rabbitmq-rpc-server


http://chatgpt.dhexx.cn/article/ccJAoYyV.shtml

相关文章

springcloud 远程rpc调用接口

springcloud 远程rpc调用接口

springcloud 如果想要从注册中心中调用已经注册的接口&#xff0c;需要用如下方法。 我们现在已经将一个微服务注册到了注册中心。端口是8083的那个 并且里面有个member的RequestMapping 首先&#xff0c;创建一个微服务&#xff0c;将需要的依赖放入pom.xml和配置好apppli…
阅读更多...
浅谈RPC调用过程

浅谈RPC调用过程

RPC(Remote Procedure Call) - 远程过程调用&#xff0c;是一个计算机通信协议&#xff0c;它允许运行于一台计算机的程序调用另一台计算机的子程序&#xff0c;而无需额外地为这个交互作用编程。RPC主要应用在分布式系统架构中不同的系统之间的远程通信和相互调用。 举个例子…
阅读更多...
远程RPC调用用于js逆向

远程RPC调用用于js逆向

这里只是记录于学习&#xff0c;资源来自于大佬的github&#xff0c;有十分详细的介绍&#xff0c;我仅仅拿来一用。https://github.com/jxhczhl/JsRpc。&#xff08;注RPC在某些网站会被检测到&#xff09; 在右侧下载适合自己系统的exe文件&#xff0c;双击打开 练习对象是某…
阅读更多...
使用 webservice 实现 RPC 调用

使用 webservice 实现 RPC 调用

WebService 介绍 Web service 是一个平台独立的&#xff0c;低耦合的 web 的应用程序用于开发分布式的互操作的应用程序。Web Service 技术&#xff0c; 能使得运行在不同机器上的不同应用无须借助附加的、专门的第三方软件或硬件&#xff0c; 就可相互交换数据或集成。 SO…
阅读更多...
RPC调用链通信方法

RPC调用链通信方法

RPC调用链可以将远程过程调用变成一个有效的堆栈。 当我们编写应用程序时&#xff0c;我们中的许多人认为REST是服务间通信的一个通用标准。然而&#xff0c;还有许多其他形式的通信&#xff0c;RPC就是其中之一。值得注意的是&#xff0c;通信方式的选择取决于许多因素&#…
阅读更多...
使用Dubbo实现简单的RPC调用（Spring配置文件版）

使用Dubbo实现简单的RPC调用（Spring配置文件版）

更多文章和资源欢迎访问&#xff1a;SuperCoder Blog 文章目录 更多文章和资源欢迎访问&#xff1a;[SuperCoder Blog](https://supercoder.com.cn)前言一、Dubbo架构二、安装zookeeper三、代码实战1. 创建服务提供方2. 创建服务消费方 四、服务调用 前言 Dubbo是阿里巴巴公司…
阅读更多...
用一个简单的函数来理一下RPC调用过程

用一个简单的函数来理一下RPC调用过程

1.什么是RPC RPC&#xff08;Remote Procedure Call&#xff09;远程过程调度&#xff0c;简单的理解就是一个节点请求另一个节点的提供的服务。 2.远程调用要面临的三个问题。 &#xff08;1&#xff09;Call ID映射。本地调用中&#xff0c;函数体是直接通过函数指针来指定…
阅读更多...
Dubbo的RPC调用流程

Dubbo的RPC调用流程

首先在客户端启动时会从注册中心拉去和订阅对应的服务列表,Cluster会把拉取到的服务列表聚合成一个cluster&#xff0c;每次RPC调用前会通过Directory#list获取providers地址(已经生成好的invoker列表)&#xff0c;获取这些服务列表给后续路由和负载均衡使用&#xff0c;框架内…
阅读更多...
RPC 调用原理

RPC 调用原理

RPC调用原理 RPC原理 RPC(Remote Procedure Call)即远程过程调用&#xff0c;允许一台计算机调用另一台计算机上的程序得到结果&#xff0c;而代码中不需要做额外的编程&#xff0c;就像在本地调用一样。分布式的应用可以借助RPC来完成服务之间的调用。 RPC框架原理 在RPC框架…
阅读更多...
RPC调用完整流程

RPC调用完整流程

RPC调用完整流程&#xff1a;   调用方持续把请求参数对象序列化成二进制数据&#xff0c;经过 TCP 传输到服务提供方&#xff1b;服务提供方从 TCP 通道里面接收到二进制数据&#xff1b;根据 RPC 协议&#xff0c;服务提供方将二进制数据分割出不同的请求数据&#xff0c;经…
阅读更多...
RPC调用

RPC调用

1.RPC调用、本地调用和HTTP调用 网络调用一般是基于HTTP协议进行调用&#xff0c;RPC是使用TCP或者UDP协议进行调用&#xff0c;效率上回更好&#xff1b; 常用的RPC框架有Thrift、GRPC等&#xff1b; 2.RPC的调用时序图 1&#xff09;调用方调用Proxy请求服务&#xff0c;并…
阅读更多...
简单实现rpc调用

简单实现rpc调用

rpc调用 代码地址 查看代码 rpc称远程调用过程&#xff0c;在rpc调用过程中需要解决的几个问题。 代理协议序列化 技术选型 协议&#xff0c;我们使用java的socket编程套接字代理&#xff0c;使用java的动态代理序列化&#xff0c;使用java的原生的序列化。 rpc衍生出来…
阅读更多...
Go rpc调用

Go rpc调用

RPC原理图 客户端Stub: 客户端存根服务端Stub: 服务端存根 RPC技术架构 服务端&#xff1a;server.go package mainimport ("math""net""net/http""net/rpc" )type MathUtil struct { }// 该方法向外暴露&#xff1a;提供计算圆形面…
阅读更多...
rpc调用 java_RPC调用的简单实现

rpc调用 java_RPC调用的简单实现

RPC调用流程 流程描述： 1.服务调用者发送请求(interface#method#args) 2.客户端进行StringEncode编码 3.数据写到服务提供者 4.服务提供者接受请求 5.将接收的包进行StringDecode解码 6.服务提供方调用对应api 7.服务提供方响应方法调用结果 8.服务提供方将结果集进行StringEn…
阅读更多...
RPC调用和HTTP调用的区别

RPC调用和HTTP调用的区别

一.远程调用方式 无论是微服务还是分布式服务&#xff08;都是SOA&#xff0c;都是面向服务编程&#xff09;&#xff0c;都面临着服务间的远程调用。那么服务间的远程调用方式有哪些呢&#xff1f; 常见的远程调用方式有以下几种&#xff1a; RPC&#xff1a;Remote Produce…
阅读更多...
RPC远程调用通俗理解

RPC远程调用通俗理解

先从一个案例来讲RPC 为了提升饭店的服务能力&#xff0c;饭店从一开始只有一个负责所有事情的厨师发展成有厨师、切菜师、备菜师等多个角色。在饭店只有一个厨师的时候&#xff0c;厨师想要做出一道美味的番茄炒蛋的时候&#xff0c;他需要自己洗番茄、切番茄、打鸡蛋、炒菜。…
阅读更多...
RPC通信基本原理 -- 浅析RPC远程过程调用基本原理

RPC通信基本原理 -- 浅析RPC远程过程调用基本原理

一、RPC基本概念 1.1、RPC简介 RPC 的全称是 Remote Procedure Call是一种进程间通信方式。RPC只是一个概念 而不是具体的协议或框架。它允许程序调用另一个地址空间&#xff08;通常是共享网络的另一台机器上&#xff09;的过程或函数&#xff0c;而不用程序员显式编码这个远…
阅读更多...
RPC服务调用

RPC服务调用

RPC服务调用 一、RPC实现原理深入分析1、RPC的定义&#xff08;1&#xff09;RPC作用&#xff08;2&#xff09;RPC核心组成&#xff08;3&#xff09;RPC调用方式 二、精简版RPC调用代码实现1、场景2、接口设计3、序列化协议 最近在做调用系统相关的服务&#xff0c;采用主从的…
阅读更多...
远程过程调用(RPC)详解

远程过程调用(RPC)详解

本文介绍了什么是远程过程调用(RPC)&#xff0c;RPC 有哪些常用的方法&#xff0c;RPC 经历了哪些发展阶段&#xff0c;以及比较了各种 RPC 技术的优劣。 什么是 RPC RPC 是远程过程调用&#xff08;Remote Procedure Call&#xff09;的缩写形式&#xff0c;Birrell 和 Nels…
阅读更多...
Simple Schnorr Multi-Signatures with Applications to Bitcoin 学习笔记

Simple Schnorr Multi-Signatures with Applications to Bitcoin 学习笔记

1. 引言 Blockstream团队2018年论文《Simple Schnorr Multi-Signatures with Applications to Bitcoin》。 对应的代码实现&#xff1a; https://github.com/KZen-networks/multi-party-schnorr https://github.com/lovesh/signature-schemes 论文要点&#xff1a; MuSig——…
阅读更多...
推荐文章

Copyright @ 2022~2023