从零写分布式RPC框架系列 1.0 （3）RPC-Server模块设计实现

深碍√TFBOYSˉ_ 2022-04-17 21:11 201阅读 0赞

RPC-Server模块负责（1）将@RpcService注解标记的服务和自身信息注册到ZK集群，（2）对外提供RPC服务实现，处理来自RPC-Client的请求。该模块整体的核心类为 RpcServer ,而真正处理请求的核心类是 RpcServerHandler 。另外还有一个 ZKServiceRegistry 负责和 ZK集群交互。

## 系列文章： ##

项目GitHub地址：[https://github.com/linshenkx/rpc-netty-spring-boot-starter][https_github.com_linshenkx_rpc-netty-spring-boot-starter]

[从零写分布式RPC框架 系列 1.0 （1）架构设计][RPC_ _ 1.0 _1]  
[从零写分布式RPC框架 系列 1.0 （2）RPC-Common模块设计实现][RPC_ _ 1.0 _2_RPC-Common]  
[从零写分布式RPC框架 系列 1.0 （3）RPC-Server模块设计实现][RPC_ _ 1.0 _3_RPC-Server]  
[从零写分布式RPC框架 系列 1.0 （4）RPC-Client模块设计实现][RPC_ _ 1.0 _4_RPC-Client]  
[从零写分布式RPC框架 系列 1.0 （5）整合测试][RPC_ _ 1.0 _5]  
[使用gpg插件发布jar包到Maven中央仓库 完整实践][gpg_jar_Maven_]

### 文章目录 ###

*   *  系列文章：
     *  一 介绍
     *   *  1 整体结构
         *  2 模块介绍
     *  二 pom文件
     *  三 简单组件：@RpcService和属性注入类
     *  四 ZKServiceRegistry
     *   *  init方法
         *  register方法
     *  五 RpcServer
     *   *  setApplicationContext方法（服务扫描）
         *  afterPropertiesSet方法（服务启动、服务注册）
     *  六 RpcServerHandler
     *   *  channelRead0 方法
         *  handler方法
     *  七 RpcServerAutoConfiguration 和 spring.factories
     *   *  1 spring.factories
         *  2 RpcServerAutoConfiguration

## 一 介绍 ##

### 1 整体结构 ###

![项目结构][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]

### 2 模块介绍 ###

注意，因为最终是以 spring-boot-starter 的形式对外提供，所以我把过程命名为 spring-boot-autoconfigure 的格式，再用一个spring-boot-starter对其包装。  
整体结构如下

1.  @RpcService注解  
    用于标注 Rpc 服务实现类，其value为 Class<?> 类型，RpcSever类启动的时候将扫描所有@RpcService标记类，并根据其value获取其 Rpc实现。
2.  RpcServerHandler  
    Rpc服务端处理器，将嵌入Netty 分配的管道流中，并利用反射技术处理来自客户端的RpcRequest生成结果封装成RpcResponse返回给客户端。
3.  RpcServerProperties和ZKProperties  
    这两个类是属性注入类，都使用了@ConfigurationProperties注解。
4.  ZKServiceRegistry  
    主要负责 连接ZK集群 和 将服务信息和自身服务地址注册到ZK集群 中。
5.  RpcServer  
    RPC-Server模块核心类，负责 管理@RpcService标记类 和 启动RPC服务。
6.  RpcServerAutoConfiguration和spring.factories文件  
    封装成spring-boot-starter所需配置，开启以上各类基于spring的自动装配。

## 二 pom文件 ##

spring-boot-configuration-processor用于注入配置属性  
zkclient提供和ZK集群交互的能力  
netty-all结合[rpc-netty-common][RPC_ _ 1.0 _2_RPC-Common]中的组件即可提供Netty服务。

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>rpc-netty-server-spring-boot-autoconfigure</artifactId>
    
        <parent>
            <groupId>com.github.linshenkx</groupId>
            <artifactId>rpc-netty-spring-boot-starter</artifactId>
            <version>1.0.5.RELEASE</version>
            <relativePath>../</relativePath>
        </parent>
    
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter</artifactId>
            </dependency>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-configuration-processor</artifactId>
            </dependency>
            <dependency>
                <groupId>com.github.linshenkx</groupId>
                <artifactId>rpc-netty-common</artifactId>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
            </dependency>
            <dependency>
                <groupId>com.101tec</groupId>
                <artifactId>zkclient</artifactId>
            </dependency>
        </dependencies>
    
    </project>

## 三 简单组件：@RpcService和属性注入类 ##

@RpcService的实现比较简单，需要注意的是 利用@Service 组合注解来将标记类收归Spring管理，借此在RpcServer可以方便实现扫描获取。注意使用时其value应该是对应的服务接口类而不是当前被标记的服务实现类。因为服务接口类才代表契约，而本地服务实现类的命名等则不受限制。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: * RPC服务注解（标注在rpc服务实现类上） * 使用@Service注解使被@RpcService标注的类都能被Spring管理 */
    @Target({ ElementType.TYPE})
    @Retention(RetentionPolicy.RUNTIME)
    @Service
    public @interface RpcService { 
        Class<?> value();
    }

属性注入类的实现比较简单，在这里可以给各个参数配置默认值。  
注意这里并没有使用@Component将其收归Spring管理，而是在需要使用对应属性注入类的时候在类上使用@EnableConfigurationProperties(RpcServerProperties.class)和再用@Autowired将其引入。这样可以由Spring确保其先于使用类实例化。

@Data
    @ConfigurationProperties(prefix = "rpc.server")
    public class RpcServerProperties { 
        private int port=9000;
    }
    
    @Data
    @ConfigurationProperties(prefix = "zk")
    public class ZKProperties { 
        private List<String> addressList = new ArrayList<>();
        private int sessionTimeOut=5000;
        private int connectTimeOut=1000;
        private String registryPath="/defaultRegistry";
    
    }

## 四 ZKServiceRegistry ##

### init方法 ###

init()方法将于ZKServiceRegistry构造完成后执行，将使用用户提供的zk地址列表随机挑选一地址进行连接。后续需进行改进，对于有多个地址的情况，不应该只尝试一次，如果随机选择到的地址刚好由于网络问题无法及时连接，则会影响项目启动，此时应该选择其他地址进行尝试。

### register方法 ###

register(String serviceName,String serviceAddress)方法将根据 **（1）配置文件的registryPath（默认为 /defaultRegistry）**\+**（2）服务名ServiceName 在zk集群生成 永久service节点**，再在永久节点下生成 临时address节点（格式为/address-递增数字），其临时节点内容为 serviceAddress。最终的节点路径形式如 ·`/defaultRegistry/com.github.linshenkx.rpclib.HelloService/address-0000000033`。在连接与ZK集群断开后，临时节点会自动移除。

由此，当有多个RPC-Server提供同一Service的时候，将在同一永久service节点下生成包含各自地址信息的临时address节点。这样，RPC-Client就可以提供查询Service节点下的子节点，获取能提供对应服务实现的RPC-Server列表，实现服务发现。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: zookeeper服务注册中心 */
    @Log4j2
    @EnableConfigurationProperties(ZKProperties.class)
    public class ZKServiceRegistry { 
    
    
      @Autowired
      private ZKProperties zkProperties;
    
      private ZkClient zkClient;
    
      @PostConstruct
      public void init() { 
        // 创建 ZooKeeper 客户端
        zkClient = new ZkClient(getAddress(zkProperties.getAddressList()), zkProperties.getSessionTimeOut(), zkProperties.getConnectTimeOut());
        log.info("connect to zookeeper");
      }
    
      public String getAddress(List<String> addressList){ 
        if(CollectionUtils.isEmpty(addressList)){ 
          String defaultAddress="localhost:2181";
          log.error("addressList is empty,using defaultAddress:"+defaultAddress);
          return defaultAddress;
        }
        //待改进策略
        String address= getRandomAddress(addressList);
        log.info("using address:"+address);
        return address;
      }
    
      private String getRandomAddress(List<String> addressList){ 
        return addressList.get(ThreadLocalRandom.current().nextInt(addressList.size()));
      }
    
      /** * 为服务端提供注册 * 将服务地址注册到对应服务名下 * 断开连接后地址自动清除 * @param serviceName * @param serviceAddress */
      public void register(String serviceName, String serviceAddress) { 
        // 创建 registry 节点（持久）
        String registryPath = zkProperties.getRegistryPath();
        if (!zkClient.exists(registryPath)) { 
          zkClient.createPersistent(registryPath);
          log.info("create registry node: {}", registryPath);
        }
        // 创建 service 节点（持久）
        String servicePath = registryPath + "/" + serviceName;
        if (!zkClient.exists(servicePath)) { 
          zkClient.createPersistent(servicePath);
          log.info("create service node: {}", servicePath);
        }
        // 创建 address 节点（临时）
        String addressPath = servicePath + "/address-";
        String addressNode = zkClient.createEphemeralSequential(addressPath, serviceAddress);
        log.info("create address node: {}", addressNode);
      }
    
    }

## 五 RpcServer ##

### setApplicationContext方法（服务扫描） ###

RpcServer实现 ApplicationContextAware 接口来获取 ApplicationContext感知能力。并在ApplicationContextAware接口带来的 setApplicationContext 方法中完成 将RpcService收归管理 的任务。需要注意的是这个方法将在类初始化完成后执行。

前文已经介绍到 @RpcService 注解提供组合@Service注解将标记类收归Spring管理，所以这里可以利用 ApplicationContext 获取 所有标记类。再以@Service的value的服务接口类的类名作为key，标记类（即服务实现类）为value存入handlerMap中，收归RpcServer管理。

### afterPropertiesSet方法（服务启动、服务注册） ###

RpcServer还实现了InitializingBean 接口来获取使用 afterPropertiesSet 方法的能力，该方法将在类初始化完成后执行，但晚于setApplicationContext方法。故执行该方法时RpcServer已完成 **服务扫描**，已在handlerMap中管理着服务实现类。

afterPropertiesSet方法的主要任务有：

1.  **服务启动**：启动RPC服务器（提供Netty连接服务）  
    其核心实现由 RpcHandler 提供
2.  **服务注册**：将服务信息和自身地址注册到注册中心（ZK集群）  
    其核心实现由 RpcServiceRegistry 提供

注意启动过程如果抛出异常将执行优雅关闭。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: TODO */
    @Log4j2
    @AutoConfigureAfter({ ZKServiceRegistry.class})
    @EnableConfigurationProperties(RpcServerProperties.class)
    public class RpcServer implements ApplicationContextAware, InitializingBean { 
    
        private Map<String,Object> handlerMap=new HashMap<>();
    
        @Autowired
        private RpcServerProperties rpcProperties;
    
        @Autowired
        private ZKServiceRegistry rpcServiceRegistry;
    
        /** * 在类初始化时执行，将所有被@RpcService标记的类纳入管理 * @param applicationContext * @throws BeansException */
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
    
            //获取带有@RpcService注解的类
            Map<String,Object> rpcServiceMap=applicationContext.getBeansWithAnnotation(RpcService.class);
            //以@RpcService注解的value的类的类名为键将该标记类存入handlerMap
            if(!CollectionUtils.isEmpty(rpcServiceMap)){ 
                for(Object object:rpcServiceMap.values()){ 
                    RpcService rpcService=object.getClass().getAnnotation(RpcService.class);
                    String serviceName=rpcService.value().getName();
                    handlerMap.put(serviceName,object);
                }
            }
    
        }
    
    
        /** * 在所有属性值设置完成后执行，负责启动RPC服务 * @throws Exception */
        @Override
        public void afterPropertiesSet() throws Exception { 
            //管理相关childGroup
            EventLoopGroup bossGroup=new NioEventLoopGroup();
            //处理相关RPC请求
            EventLoopGroup childGroup=new NioEventLoopGroup();
    
            try { 
                //启动RPC服务
                ServerBootstrap bootstrap=new ServerBootstrap();
                bootstrap.group(bossGroup,childGroup);
                bootstrap.channel(NioServerSocketChannel.class);
                bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { 
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception { 
                        ChannelPipeline pipeline=channel.pipeline();
                        //解码RPC请求
                        pipeline.addLast(new RpcDecoder(RpcRequest.class));
                        //编码RPC请求
                        pipeline.addFirst(new RpcEncoder(RpcResponse.class));
                        //处理RPC请求
                        pipeline.addLast(new RpcServerHandler(handlerMap));
                    }
                });
                //同步启动，RPC服务器启动完毕后才执行后续代码
                ChannelFuture future=bootstrap.bind(rpcProperties.getPort()).sync();
                log.info("server started,listening on {}",rpcProperties.getPort());
                //注册RPC服务地址
                String serviceAddress= InetAddress.getLocalHost().getHostAddress()+":"+rpcProperties.getPort();
                for(String interfaceName:handlerMap.keySet()){ 
                    rpcServiceRegistry.register(interfaceName,serviceAddress);
                    log.info("register service:{}=>{}",interfaceName,serviceAddress);
                }
                //释放资源
                future.channel().closeFuture().sync();
            }catch (Exception e){ 
                log.entry("server exception",e);
            }finally { 
                //关闭RPC服务
                childGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
    
        }
    }

## 六 RpcServerHandler ##

该类是处理请求的核心类。该类继承自 Netty 的 SimpleChannelInboundHandler，其传入泛型为RpcRequest。即处理对象为 来自 RPC-Client 的 RpcRequest。该类主要通过覆写 channelRead0 来对请求进行处理。

该类在构造时即获取管理服务实现类的能力。（通过在构造方法中传入handlerMap实现）

### channelRead0 方法 ###

该方法先创建一个响应对象RpcResponse，并将处理的RpcRequest的请求Id设置给它，以形成一一对应关系。再执行handle方法来获取处理结果（或异常）并设置给RpcResponse，然后将结果返回（实际上是进入下一步，由下一个ChannelHandler继续处理，在这个项目中即RpcEncoder）。

### handler方法 ###

核心中的核心。但本身并不复杂，使用动态代理技术执行目标方法得到结果而已。  
首先根据RpcRequest的InterfaceName字段获取对应的服务实现类，再从RpcRequest中获取反射调用所需的变量，如方法名、参数类型、参数列表等，最后执行反射调用即可。  
目前使用的是jdk的动态代理，以后应该加上cglib才更完整。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: RPC服务端处理器（处理RpcRequest） */
    @Log4j2
    public class RpcServerHandler extends SimpleChannelInboundHandler<RpcRequest> { 
    
    
      /** * 存放 服务名称 与 服务实例 之间的映射关系 */
      private final Map<String, Object> handlerMap;
    
      public RpcServerHandler(Map<String, Object> handlerMap) { 
        this.handlerMap = handlerMap;
      }
    
      @Override
      public void channelRead0(ChannelHandlerContext ctx, RpcRequest request) throws Exception { 
        log.info("channelRead0 begin");
        // 创建 RPC 响应对象
        RpcResponse response = new RpcResponse();
        response.setRequestId(request.getRequestId());
        try { 
          // 处理 RPC 请求成功
          Object result = handle(request);
          response.setResult(result);
        } catch (Exception e) { 
          // 处理 RPC 请求失败
          response.setException(e);
          log.error("handle result failure", e);
        }
        // 写入 RPC 响应对象（写入完毕后立即关闭与客户端的连接）
        ctx.writeAndFlush(response).addListener(ChannelFutureListener.CLOSE);
      }
    
      @Override
      public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
        log.error("server caught exception", cause);
        ctx.close();
      }
    
      private Object handle(RpcRequest request) throws Exception { 
        log.info("handle begin");
        // 获取服务实例
        String serviceName = request.getInterfaceName();
        Object serviceBean = handlerMap.get(serviceName);
        if (serviceBean == null) { 
          throw new RuntimeException(String.format("can not find service bean by key: %s", serviceName));
        }
        // 获取反射调用所需的变量
        Class<?> serviceClass = serviceBean.getClass();
        String methodName = request.getMethodName();
        log.info(methodName);
        Class<?>[] parameterTypes = request.getParameterTypes();
        log.info(parameterTypes[0].getName());
        Object[] parameters = request.getParameters();
        // 执行反射调用
        Method method = serviceClass.getMethod(methodName, parameterTypes);
        method.setAccessible(true);
        log.info(parameters[0].toString());
        return method.invoke(serviceBean, parameters);
      }
    }

## 七 RpcServerAutoConfiguration 和 spring.factories ##

这两个是封装成spring-boot-starter所需的配置，因为spring-boot默认只会扫描启动类同级目录下的注解，对于外部依赖不会扫描，除非指定扫描，但这样显然不是我们的目的。所以需要使用这两个文件开启基于spring的自动装配。

### 1 spring.factories ###

在resources/META-INF下创建spring.factories文件，指定自动装配的类，书写格式为  
`org.springframework.boot.autoconfigure.EnableAutoConfiguration=类名全称A,类名全称B`  
注意类名一定要全称，多个类用逗号隔开，换行在末尾加 `\` ，而且通过这种方式装配会有顺序，顺序与文件中声明一致。  
通过这种方式实现自动注入在类多的时候基本不可行，因为可读性太差了，而且装配顺序需要人工维护。  
所以一般是在这里装配一个自动配置类，通过自动配置类再去注入其他类，并实现更高级功能。

# AutoConfiguration
    org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
    com.github.linshenkx.rpcnettyserverspringbootautoconfigure.RpcServerAutoConfiguration

### 2 RpcServerAutoConfiguration ###

如下，可以利用 @ConditionalOnClass 避免错误装配，通过 @ConditionalOnMissingBean 提供让用户注入实现的机会。也可以在这里指定装配顺序。

/** * @version V1.0 * @author: lin_shen * @date: 2018/11/2 * @Description: TODO */
    @Configuration
    @ConditionalOnClass(RpcServer.class)
    public class RpcServerAutoConfiguration { 
        @ConditionalOnMissingBean
        @Bean
        public RpcServerProperties defaultRpcServerProperties(){ 
            return new RpcServerProperties();
        }
    
        @ConditionalOnMissingBean
        @Bean
        public ZKProperties defaultZKProperties(){ 
            return new ZKProperties();
        }
    
        @ConditionalOnMissingBean
        @Bean
        public ZKServiceRegistry zkServiceRegistry(){ 
            return new ZKServiceRegistry();
        }
    
        @Bean
        public RpcServer rpcServer(){ 
            return new RpcServer();
        }
        
    }

[https_github.com_linshenkx_rpc-netty-spring-boot-starter]: https://github.com/linshenkx/rpc-netty-spring-boot-starter
[RPC_ _ 1.0 _1]: https://blog.csdn.net/alinyua/article/details/83757901
[RPC_ _ 1.0 _2_RPC-Common]: https://blog.csdn.net/alinyua/article/details/83819288
[RPC_ _ 1.0 _3_RPC-Server]: https://blog.csdn.net/alinyua/article/details/83823907
[RPC_ _ 1.0 _4_RPC-Client]: https://blog.csdn.net/alinyua/article/details/83859468
[RPC_ _ 1.0 _5]: https://blog.csdn.net/alinyua/article/details/83863043
[gpg_jar_Maven_]: https://blog.csdn.net/alinyua/article/details/83687250
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]: /images/20220418/ac2041404dcb48bcad46ef4f51b87882.png