Simple RPC - 07 从零开始设计一个服务端(下)_RPC服务的实现

2024-08-24 19:36

本文主要是介绍Simple RPC - 07 从零开始设计一个服务端(下)_RPC服务的实现,希望对大家解决编程问题提供一定的参考价值,需要的开发者们随着小编来一起学习吧!

文章目录

  • Pre
  • RPC服务实现
    • 服务注册
    • 请求处理
  • 设计: 请求分发机制

在这里插入图片描述

Pre

Simple RPC - 01 框架原理及总体架构初探

Simple RPC - 02 通用高性能序列化和反序列化设计与实现

Simple RPC - 03 借助Netty实现异步网络通信

Simple RPC - 04 从零开始设计一个客户端(上)

Simple RPC - 05 从零开始设计一个客户端(下)_ 依赖倒置和SPI


RPC服务实现

服务端的RPC服务主要包含两个核心功能:

  1. 服务注册:将服务的实现类注册到RPC框架中。

  2. 请求处理:接收并处理来自客户端的RPC请求。


服务注册

服务注册通过一个Map<String, Object>结构来实现,其中Key为服务名,Value为服务实现类的实例。

把服务的实现类注册到 RPC 框架中,这个逻辑的实现很简单,我们只要使用一个合适的数据结构,记录下所有注册的实例就可以了,后面在处理客户端请求的时候,会用到这个数据结构来查找服务实例

@Singleton
public class RpcRequestHandler implements RequestHandler, ServiceProviderRegistry {private final Map<String, Object> serviceProviders = new ConcurrentHashMap<>();@Overridepublic synchronized <T> void addServiceProvider(Class<? extends T> serviceClass, T serviceProvider) {serviceProviders.put(serviceClass.getCanonicalName(), serviceProvider);}
}

请求处理

首先来看服务端中,使用 Netty 接收所有请求数据的处理类 RequestInvocationchannelRead0 方法

/*** 处理通道读取事件** @param channelHandlerContext 上下文处理器,用于管理通道事件* @param request 入站请求命令* @throws Exception 如果没有找到处理请求的处理器或者其他异常*/@Overrideprotected void channelRead0(ChannelHandlerContext channelHandlerContext, Command request) throws Exception {// 根据请求头的类型获取对应的请求处理器RequestHandler handler = requestHandlerRegistry.get(request.getHeader().getType());if(null != handler) {// 处理请求并获取响应Command response = handler.handle(request);if(null != response) {// 将响应写入并刷新通道上下文,并添加监听器处理写入结果channelHandlerContext.writeAndFlush(response).addListener((ChannelFutureListener) channelFuture -> {if (!channelFuture.isSuccess()) {// 如果写入响应失败,记录日志并关闭通道logger.warn("Write response failed!", channelFuture.cause());channelHandlerContext.channel().close();}});} else {// 如果响应为空,记录警告日志logger.warn("Response is null!");}} else {// 如果没有找到处理请求的处理器,抛出异常throw new Exception(String.format("No handler for request with type: %d!", request.getHeader().getType()));}}

根据请求命令的 Hdader 中的请求类型 type,去 requestHandlerRegistry 中查找对应的请求处理器 RequestHandler,然后调用请求处理器去处理请求,最后把结果发送给客户端

这种通过“请求中的类型”,把请求分发到对应的处理类或者处理方法的设计,在服务端处理请求的场景中,这是一个很常用的方法。我们这里使用的也是同样的设计,不同的是,我们使用了一个命令注册机制,让这个路由分发的过程省略了大量的 if-else 或者是 switch 代码。这样做的好处是,可以很方便地扩展命令处理器,而不用修改路由分发的方法,并且代码看起来更加优雅。

/*** 请求处理器注册类,用于管理和获取不同的请求处理器* 该类通过 Singleton 模式确保只有一个实例,* 并在类加载时初始化所有已知的请求处理器* @author artisan*/
public class RequestHandlerRegistry {private static final Logger logger = LoggerFactory.getLogger(RequestHandlerRegistry.class);/*** 存储请求处理器的映射,键为处理器类型,值为处理器实例*/private Map<Integer, RequestHandler> handlerMap = new HashMap<>();/*** Singleton 实例*/private static RequestHandlerRegistry instance = null;/*** 获取 RequestHandlerRegistry 的单例实例* 如果实例不存在,则创建一个新的实例* * @return RequestHandlerRegistry 的单例实例*/public static RequestHandlerRegistry getInstance() {if (null == instance) {instance = new RequestHandlerRegistry();}return instance;}/*** 私有构造方法,用于在实例创建时加载所有请求处理器* 通过 ServiceSupport 加载所有实现 RequestHandler 接口的实例,* 并将它们添加到 handlerMap 中*/private RequestHandlerRegistry() {Collection<RequestHandler> requestHandlers = ServiceSupport.loadAll(RequestHandler.class);for (RequestHandler requestHandler : requestHandlers) {handlerMap.put(requestHandler.type(), requestHandler);logger.info("Load request handler, type: {}, class: {}.", requestHandler.type(), requestHandler.getClass().getCanonicalName());}}/*** 根据请求类型获取对应的请求处理器* * @param type 请求处理器的类型* @return 对应的请求处理器实例,如果不存在则返回 null*/public RequestHandler get(int type) {return handlerMap.get(type);}
}

接下来,我们看下通过RpcRequestHandler处理具体的客户端RPC请求

 /*** 处理请求命令的方法* 该方法的主要职责是解析请求命令,查找并调用相应服务,然后返回处理结果* * @param requestCommand 请求命令,包含服务调用信息和参数* @return 返回命令,包含调用结果或错误信息*/@Overridepublic Command handle(Command requestCommand) {Header header = requestCommand.getHeader();// 从payload中反序列化RpcRequest,获取服务调用请求的具体信息RpcRequest rpcRequest = SerializeSupport.parse(requestCommand.getPayload());try {// 根据rpcRequest中的服务名,查找已注册的服务提供方Object serviceProvider = serviceProviders.get(rpcRequest.getInterfaceName());if(serviceProvider != null) {// 服务提供者存在,反序列化参数并获取具体方法进行调用String arg = SerializeSupport.parse(rpcRequest.getSerializedArguments());// 通过反射调用服务方法Method method = serviceProvider.getClass().getMethod(rpcRequest.getMethodName(), String.class);String result = (String ) method.invoke(serviceProvider, arg);// 将调用结果序列化并封装成响应命令返回return new Command(new ResponseHeader(type(), header.getVersion(), header.getRequestId()), SerializeSupport.serialize(result));}// 未找到对应的服务提供方,返回错误响应logger.warn("No service Provider of {}#{}(String)!", rpcRequest.getInterfaceName(), rpcRequest.getMethodName());return new Command(new ResponseHeader(type(), header.getVersion(), header.getRequestId(), Code.NO_PROVIDER.getCode(), "No provider!"), new byte[0]);} catch (Throwable t) {// 处理过程中发生异常,记录并返回错误响应logger.warn("Exception: ", t);return new Command(new ResponseHeader(type(), header.getVersion(), header.getRequestId(), Code.UNKNOWN_ERROR.getCode(), t.getMessage()), new byte[0]);}}

核心流程如下:

  1. requestCommandpayload 属性反序列化成为 RpcRequest
  2. 根据 rpcRequest 中的服务名,去成员变量 serviceProviders 中查找已注册服务实现类的实例;
  3. 找到服务提供者之后,利用 Java 反射机制调用服务的对应方法;
  4. 把结果封装成响应命令并返回,在 RequestInvocation 中,它会把这个响应命令发送给客户端。

再来看成员变量 serviceProviders,它的定义是:Map<String/service name/, Object/service provider/> serviceProviders

它实际上就是一个 Map,Key 就是服务名,Value 就是服务提供方,也就是服务实现类的实例。这个 Map 的数据从哪儿来的呢?

我们来看一下 RpcRequestHandler 这个类的定义

@Singleton
public class RpcRequestHandler implements RequestHandler, ServiceProviderRegistry {/*** 同步地向服务提供者集合中添加一个新的服务提供者* 此方法为类的公共接口一部分,提供了一种向内部服务提供者映射添加新服务提供者的方式** @param serviceClass 服务接口类,指定了服务提供者的类型期望* @param serviceProvider 实际的服务提供者实例,实现了指定的服务接口** 选择使用同步方法以确保线程安全,因为在多线程环境中,* 可能会有多个线程尝试同时向服务提供者集合中添加服务** 使用服务提供者的类的规范名称(canonical name)作为键进行存储,是为了确保* 通过类名唯一标识服务提供者,避免了由于类加载器差异导致的潜在问题** 记录日志是为了在系统运行时能够追踪到服务提供者的添加操作,有助于调试和系统维护*/@Overridepublic synchronized <T> void addServiceProvider(Class<? extends T> serviceClass, T serviceProvider) {serviceProviders.put(serviceClass.getCanonicalName(), serviceProvider);logger.info("Add service: {}, provider: {}.",serviceClass.getCanonicalName(),serviceProvider.getClass().getCanonicalName());}}

这个类不仅实现了处理客户端请求的 RequestHandler 接口,同时还实现了注册 RPC 服务 ServiceProviderRegistry 接口,也就是说,RPC 框架服务端需要实现的两个功能——注册 RPC 服务和处理客户端 RPC 请求,都是在这一个类 RpcRequestHandler 中实现的,所以说,这个类是这个 RPC 框架服务端最核心的部分。

成员变量 serviceProviders 这个 Map 中的数据,也就是在 addServiceProvider 这个方法的实现中添加进去的


@Singleton

RpcRequestHandler 上增加了一个注解 @Singleton,限定这个类它是一个单例模式,这样确保在进程中任何一个地方,无论通过 ServiceSupport 获取 RequestHandler 或者 ServiceProviderRegistry 这两个接口的实现类,拿到的都是 RpcRequestHandler 这个类的唯一的一个实例。 实现如下


/*** 提供服务加载功能的支持类,特别是处理单例服务* @author artisan*/
public class ServiceSupport {/*** 存储单例服务的映射,确保每个服务只有一个实例*/private final static Map<String, Object> singletonServices = new HashMap<>();/*** 加载单例服务实例** @param service 服务类的Class对象* @param <S> 服务类的类型参数* @return 单例服务实例* @throws ServiceLoadException 如果找不到服务实例*/public synchronized static <S> S load(Class<S> service) {return StreamSupport.stream(ServiceLoader.load(service).spliterator(), false).map(ServiceSupport::singletonFilter).findFirst().orElseThrow(ServiceLoadException::new);}/*** 加载所有服务实例** @param service 服务类的Class对象* @param <S> 服务类的类型参数* @return 所有服务实例的集合*/public synchronized static <S> Collection<S> loadAll(Class<S> service) {return StreamSupport.stream(ServiceLoader.load(service).spliterator(), false).map(ServiceSupport::singletonFilter).collect(Collectors.toList());}/*** 对服务实例进行单例过滤** @param service 服务实例* @param <S> 服务类的类型参数* @return 单例过滤后的服务实例,如果该服务是单例的并且已有实例存在,则返回已存在的实例*/@SuppressWarnings("unchecked")private static <S> S singletonFilter(S service) {if(service.getClass().isAnnotationPresent(Singleton.class)) {String className = service.getClass().getCanonicalName();Object singletonInstance = singletonServices.putIfAbsent(className, service);return singletonInstance == null ? service : (S) singletonInstance;} else {return service;}}
}

设计: 请求分发机制

请求分发分为两个层次:

  1. 网络传输层:通过RequestInvocation类,根据请求类型分发到对应的请求处理器。

    根据请求命令中的请求类型 (command.getHeader().getType()),分发到对应的请求处理器 RequestHandler 中

  2. 业务逻辑层:通过RpcRequestHandler类,根据服务名分发到具体的服务实现类。

    根据 RPC 请求中的服务名,把 RPC 请求分发到对应的服务实现类的实例中去

这种分层设计的目的在于保持系统的松耦合高内聚,确保网络传输与业务逻辑的清晰分离。

在这里插入图片描述

这篇关于Simple RPC - 07 从零开始设计一个服务端(下)_RPC服务的实现的文章就介绍到这儿,希望我们推荐的文章对编程师们有所帮助!



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

相关文章

不懂推荐算法也能设计推荐系统

本文以商业化应用推荐为例,告诉我们不懂推荐算法的产品,也能从产品侧出发, 设计出一款不错的推荐系统。 相信很多新手产品,看到算法二字,多是懵圈的。 什么排序算法、最短路径等都是相对传统的算法(注:传统是指科班出身的产品都会接触过)。但对于推荐算法,多数产品对着网上搜到的资源,都会无从下手。特别当某些推荐算法 和 “AI”扯上关系后,更是加大了理解的难度。 但,不了解推荐算法,就无法做推荐系

hdu1043(八数码问题,广搜 + hash(实现状态压缩) )

利用康拓展开将一个排列映射成一个自然数,然后就变成了普通的广搜题。 #include<iostream>#include<algorithm>#include<string>#include<stack>#include<queue>#include<map>#include<stdio.h>#include<stdlib.h>#include<ctype.h>#inclu

【C++】_list常用方法解析及模拟实现

相信自己的力量,只要对自己始终保持信心,尽自己最大努力去完成任何事,就算事情最终结果是失败了,努力了也不留遗憾。💓💓💓 目录   ✨说在前面 🍋知识点一:什么是list? •🌰1.list的定义 •🌰2.list的基本特性 •🌰3.常用接口介绍 🍋知识点二:list常用接口 •🌰1.默认成员函数 🔥构造函数(⭐) 🔥析构函数 •🌰2.list对象

【Prometheus】PromQL向量匹配实现不同标签的向量数据进行运算

✨✨ 欢迎大家来到景天科技苑✨✨ 🎈🎈 养成好习惯,先赞后看哦~🎈🎈 🏆 作者简介:景天科技苑 🏆《头衔》:大厂架构师,华为云开发者社区专家博主,阿里云开发者社区专家博主,CSDN全栈领域优质创作者,掘金优秀博主,51CTO博客专家等。 🏆《博客》:Python全栈,前后端开发,小程序开发,人工智能,js逆向,App逆向,网络系统安全,数据分析,Django,fastapi

让树莓派智能语音助手实现定时提醒功能

最初的时候是想直接在rasa 的chatbot上实现,因为rasa本身是带有remindschedule模块的。不过经过一番折腾后,忽然发现,chatbot上实现的定时,语音助手不一定会有响应。因为,我目前语音助手的代码设置了长时间无应答会结束对话,这样一来,chatbot定时提醒的触发就不会被语音助手获悉。那怎么让语音助手也具有定时提醒功能呢? 我最后选择的方法是用threading.Time

Android实现任意版本设置默认的锁屏壁纸和桌面壁纸(两张壁纸可不一致)

客户有些需求需要设置默认壁纸和锁屏壁纸  在默认情况下 这两个壁纸是相同的  如果需要默认的锁屏壁纸和桌面壁纸不一样 需要额外修改 Android13实现 替换默认桌面壁纸: 将图片文件替换frameworks/base/core/res/res/drawable-nodpi/default_wallpaper.*  (注意不能是bmp格式) 替换默认锁屏壁纸: 将图片资源放入vendo

C#实战|大乐透选号器[6]:实现实时显示已选择的红蓝球数量

哈喽,你好啊,我是雷工。 关于大乐透选号器在前面已经记录了5篇笔记,这是第6篇; 接下来实现实时显示当前选中红球数量,蓝球数量; 以下为练习笔记。 01 效果演示 当选择和取消选择红球或蓝球时,在对应的位置显示实时已选择的红球、蓝球的数量; 02 标签名称 分别设置Label标签名称为:lblRedCount、lblBlueCount

【区块链 + 人才服务】可信教育区块链治理系统 | FISCO BCOS应用案例

伴随着区块链技术的不断完善,其在教育信息化中的应用也在持续发展。利用区块链数据共识、不可篡改的特性, 将与教育相关的数据要素在区块链上进行存证确权,在确保数据可信的前提下,促进教育的公平、透明、开放,为教育教学质量提升赋能,实现教育数据的安全共享、高等教育体系的智慧治理。 可信教育区块链治理系统的顶层治理架构由教育部、高校、企业、学生等多方角色共同参与建设、维护,支撑教育资源共享、教学质量评估、

uva 10014 Simple calculations(数学推导)

直接按照题意来推导最后的结果就行了。 开始的时候只做到了第一个推导,第二次没有继续下去。 代码: #include<stdio.h>int main(){int T, n, i;double a, aa, sum, temp, ans;scanf("%d", &T);while(T--){scanf("%d", &n);scanf("%lf", &first);scanf

Kubernetes PodSecurityPolicy:PSP能实现的5种主要安全策略

Kubernetes PodSecurityPolicy:PSP能实现的5种主要安全策略 1. 特权模式限制2. 宿主机资源隔离3. 用户和组管理4. 权限提升控制5. SELinux配置 💖The Begin💖点点关注,收藏不迷路💖 Kubernetes的PodSecurityPolicy(PSP)是一个关键的安全特性,它在Pod创建之前实施安全策略,确保P