从零写分布式RPC框架系列 2.0 （3）RPC-Server和RPC-Client模块改造

我就是我 2022-04-14 03:11 391阅读 0赞

2.0版本RPC-Server改动不大，主要变化在于RPC-Client使用了服务地址缓存，并引入监控机制，第一时间获取zk集群中服务地址信息变化并刷新本地缓存。另外，RPC-Client还使用了RpcClientProperties开放对负载均衡策略和序列化策略的选择。

## 系列文章： ##

专栏：[从零开始写分布式RPC框架][RPC]  
项目GitHub地址：[https://github.com/linshenkx/rpc-netty-spring-boot-starter][https_github.com_linshenkx_rpc-netty-spring-boot-starter]

[手写通用类型负载均衡路由引擎（含随机、轮询、哈希等及其带权形式）][Link 1]  
[实现 序列化引擎（支持 JDK默认、Hessian、Json、Protostuff、Xml、Avro、ProtocolBuffer、Thrift等序列化方式）][_ JDK_Hessian_Json_Protostuff_Xml_Avro_ProtocolBuffer_Thrift]  
[从零写分布式RPC框架 系列 2.0 （1）架构升级][RPC_ _ 2.0 _1]  
[从零写分布式RPC框架 系列 2.0 （2）RPC-Common模块设计实现][RPC_ _ 2.0 _2_RPC-Common]  
[从零写分布式RPC框架 系列 2.0 （3）RPC-Server和RPC-Client模块改造][RPC_ _ 2.0 _3_RPC-Server_RPC-Client]  
[从零写分布式RPC框架 系列 2.0 （4）使用BeanPostProcessor实现自定义@RpcReference注解注入][RPC_ _ 2.0 _4_BeanPostProcessor_RpcReference]

### 文章目录 ###

*   *  系列文章：
     *  RPC-Server
     *   *  1 结构图
         *  2 RpcService注解
         *  3 ServiceInfo
         *  4 RpcServer
     *  RPC-Client
     *   *  1 结构图
         *  2 RpcClientProperties
         *  3 ZKServiceDiscovery
         *  4 RpcClient

## RPC-Server ##

### 1 结构图 ###

![结构图][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]注意，RpcService注解移动到了RPC-Common模块下，另外新加了ServiceInfo代表将存到注册中心的服务信息（也在RPC-Common模块下），其他的除了RpcServer基本没有变化

### 2 RpcService注解 ###

主要是多了weight和workerThreads，分别代表权重和最大工作线程数。

/** * @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();
        int weight() default 1;
        int workerThreads() default 10;
    }

### 3 ServiceInfo ###

/** * @version V1.0 * @author: lin_shen * @date: 18-11-13 * @Description: 服务信息,用于存储到注册中心 */
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class ServiceInfo implements WeightGetAble { 
    
        private String host;
        private int port;
        /** * 权重信息 */
        private int weight;
        /** * 最大工作线程数 */
        private int workerThreads;
    
        public ServiceInfo (ServiceInfo serviceInfo){ 
            this.host = serviceInfo.host;
            this.port = serviceInfo.port;
            this.weight = serviceInfo.weight;
            this.workerThreads = serviceInfo.workerThreads;
        }
    
        @Override
        public int getWeightFactors() { 
            return getWeight();
        }
    }

### 4 RpcServer ###

RpcServer主要是多了对 serviceSemaphoreMap 和 serviceRpcServiceMap的管理。其中serviceSemaphoreMap 将作为参数传入RpcServerHandler提供限流信息，而serviceRpcServiceMap将注册到ZK集群。

/** * @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<>();
    
        /** * 存放 服务名称 与 信号量 之间的映射关系 * 用于限制每个服务的工作线程数 */
        private Map<String, Semaphore> serviceSemaphoreMap=new HashMap<>();
    
        /** * 存放 服务名称 与 服务信息 之间的映射关系 */
        private Map<String, RpcService> serviceRpcServiceMap=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和serviceSemaphoreMap
            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);
                    serviceSemaphoreMap.put(serviceName,new Semaphore(rpcService.workerThreads()));
                    serviceRpcServiceMap.put(serviceName,rpcService);
                }
            }
    
        }
    
    
        /** * 在所有属性值设置完成后执行，负责启动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.option(ChannelOption.SO_BACKLOG,1024)
                        .childOption(ChannelOption.TCP_NODELAY,true)
                        .handler(new LoggingHandler(LogLevel.INFO));
                bootstrap.childHandler(new ChannelInitializer<SocketChannel>() { 
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception { 
                        ChannelPipeline pipeline=channel.pipeline();
                        //解码RPC请求
                        pipeline.addLast(new RemotingTransporterDecoder());
                        //编码RPC请求
                        pipeline.addFirst(new RemotingTransporterEncoder());
                        //处理RPC请求
                        pipeline.addLast(new RpcServerHandler(handlerMap,serviceSemaphoreMap));
                    }
                });
                //同步启动，RPC服务器启动完毕后才执行后续代码
                ChannelFuture future=bootstrap.bind(rpcProperties.getPort()).sync();
                log.info("server started,listening on {}",rpcProperties.getPort());
    
                //启动后注册服务
                registry();
    
                //释放资源
                future.channel().closeFuture().sync();
            }catch (Exception e){ 
                log.entry("server exception",e);
            }finally { 
                //关闭RPC服务
                childGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
    
        }
    
        private void registry() throws UnknownHostException { 
            //注册RPC服务地址
            String hostAddress=InetAddress.getLocalHost().getHostAddress();
            int port=rpcProperties.getPort();
    
            for(String interfaceName:handlerMap.keySet()){ 
                ServiceInfo serviceInfo=
                        new ServiceInfo(hostAddress,port,serviceRpcServiceMap.get(interfaceName).weight(),serviceRpcServiceMap.get(interfaceName).workerThreads());
                String serviceInfoString= JSON.toJSONString(serviceInfo);
                rpcServiceRegistry.register(interfaceName,serviceInfoString);
                log.info("register service:{}=>{}",interfaceName,serviceInfoString);
            }
        }
    }

## RPC-Client ##

### 1 结构图 ###

![RPC-Client][]  
新增RpcClientProperties提供配置属性读入（路由策略和序列化方式），ZKServiceDiscovery增加ConcurrentMap<String,List> servicePathsMap来管理服务地址列表。RpcClient相应作出调整。

### 2 RpcClientProperties ###

注意这里属性用的是枚举类型而不是字符串，另外默认路由策略是随机，默认序列化策略是json

@Data
    @ConfigurationProperties(prefix = "rpc.client")
    public class RpcClientProperties { 
        private RouteStrategyEnum routeStrategy= RouteStrategyEnum.Random;
        private SerializeTypeEnum serializeType=SerializeTypeEnum.JSON;
    }

### 3 ZKServiceDiscovery ###

这里使用了IZkChildListener 来对目标路径下子节点变化进行监控，如果发生变化（新增或删减）则重新执行discover方法拉取最新服务地址列表。  
zkChildListenerMap的作用是管理服务和对应的服务地址列表监听器，避免重复注册监听器。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: zookeeper服务注册中心 */
    @Component
    @Log4j2
    @EnableConfigurationProperties(ZKProperties.class)
    public class ZKServiceDiscovery { 
    
      @Autowired
      private ZKProperties zkProperties;
    
      /** * 服务名和服务地址列表的Map */
      private ConcurrentMap<String,List<String>> servicePathsMap=new ConcurrentHashMap<>();
    
      /** * 服务监听器 Map,监听子节点服务信息 */
      private ConcurrentMap<String, IZkChildListener> zkChildListenerMap=new ConcurrentHashMap<>();
    
      private ZkClient zkClient;
    
      @PostConstruct
      public void init() { 
        // 创建 ZooKeeper 客户端
        zkClient = new ZkClient(zkProperties.getAddress(), zkProperties.getSessionTimeOut(), zkProperties.getConnectTimeOut());
        log.info("connect to zookeeper");
      }
    
      /** * * 根据服务名获取服务地址并保持监控 * @param serviceName * @return */
      public void discover(String serviceName){ 
        log.info("discovering:"+serviceName);
        String servicePath=zkProperties.getRegistryPath()+"/"+serviceName;
        //找不到对应服务
        if(!zkClient.exists(servicePath)){ 
          throw new RuntimeException("can not find any service node on path: "+servicePath);
        }
        //获取服务地址列表
        List<String> addressList=zkClient.getChildren(servicePath);
        if(CollectionUtils.isEmpty(addressList)){ 
          throw new RuntimeException("can not find any address node on path: "+servicePath);
        }
        //保存地址列表
        List<String> paths=new ArrayList<>(addressList.size());
        for(String address:addressList){ 
          paths.add(zkClient.readData(servicePath+"/"+address));
        }
        servicePathsMap.put(serviceName,paths);
        //保持监控
        if(!zkChildListenerMap.containsKey(serviceName)){ 
          IZkChildListener iZkChildListener= (parentPath, currentChilds) -> { 
            //当子节点列表变化时重新discover
            discover(serviceName);
            log.info("子节点列表发生变化 ");
          };
          zkClient.subscribeChildChanges(servicePath, iZkChildListener);
          zkChildListenerMap.put(serviceName,iZkChildListener);
        }
      }
    
      public List<String> getAddressList(String serviceName){ 
          List<String> addressList=servicePathsMap.get(serviceName);
          if(addressList==null||addressList.isEmpty()){ 
              discover(serviceName);
              return servicePathsMap.get(serviceName);
          }
          return addressList;
      }
    
    }

### 4 RpcClient ###

主要是配合RemotingTransporter做了调整和升级，整体变化不大。  
另外一个要注意的就是 ConcurrentMap<String,RouteStrategy> serviceRouteStrategyMap，用于在使用轮询策略时，为不同的服务调用保管对应的轮询器（轮询器内部存储index记录，是有状态的）。

@Log4j2
    @Component
    @AutoConfigureAfter(ZKServiceDiscovery.class)
    @EnableConfigurationProperties(RpcClientProperties.class)
    public class RpcClient { 
    
        @Autowired
        private ZKServiceDiscovery zkServiceDiscovery;
    
        @Autowired
        private RpcClientProperties rpcClientProperties;
    
        /** * 维持服务的 轮询 路由状态 * 不同服务状态不同（服务列表也不同） * 非轮询无需维持状态 */
        private ConcurrentMap<String,RouteStrategy> serviceRouteStrategyMap=new ConcurrentHashMap<>();
    
        /** * 存放请求编号与响应对象的映射关系 */
        private ConcurrentMap<Long, RemotingTransporter> remotingTransporterMap=new ConcurrentHashMap<>();
    
        @SuppressWarnings("unchecked")
        public <T> T create(final Class<?> interfaceClass){ 
            //创建动态代理对象
            return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                    new Class<?>[]{ interfaceClass},
                    (proxy, method, args) -> { 
                        //创建RPC请求对象
                        RpcRequest rpcRequest=new RpcRequest();
                        rpcRequest.setInterfaceName(method.getDeclaringClass().getName());
                        rpcRequest.setMethodName(method.getName());
                        rpcRequest.setParameterTypes(method.getParameterTypes());
                        rpcRequest.setParameters(args);
                        //获取RPC服务信息列表
                        String serviceName=interfaceClass.getName();
                        List<String> addressList=zkServiceDiscovery.getAddressList(serviceName);
    
                        List<ServiceInfo> serviceInfoList=new ArrayList<>(addressList.size());
                        for(String serviceInfoString:addressList){ 
                            serviceInfoList.add(JSON.parseObject(serviceInfoString,ServiceInfo.class));
                        }
                        //根据配置文件获取路由策略
                        log.info("使用负载均衡策略："+rpcClientProperties.getRouteStrategy());
                        log.info("使用序列化策略："+rpcClientProperties.getSerializeType());
                        RouteStrategy routeStrategy ;
                        //如果使用轮询，则需要保存状态（按服务名保存）
                        if(RouteStrategyEnum.Polling==rpcClientProperties.getRouteStrategy()){ 
                            routeStrategy=serviceRouteStrategyMap.getOrDefault(serviceName,RouteEngine.queryClusterStrategy(RouteStrategyEnum.Polling));
                            serviceRouteStrategyMap.put(serviceName,routeStrategy);
                        }else { 
                            routeStrategy= RouteEngine.queryClusterStrategy(rpcClientProperties.getRouteStrategy());
                        }
                        //根据路由策略选取服务提供方
                        ServiceInfo serviceInfo = routeStrategy.select(serviceInfoList);
    
                        RemotingTransporter remotingTransporter=new RemotingTransporter();
                        //设置flag为请求，双路，非ping，非其他，序列化方式为 配置文件中SerializeTypeEnum对应的code
                        remotingTransporter.setFlag(new RemotingTransporter.Flag(true,true,false,false,rpcClientProperties.getSerializeType().getCode()));
    
                        remotingTransporter.setBodyContent(rpcRequest);
    
                        log.info("get serviceInfo:"+serviceInfo);
                        //从RPC服务地址中解析主机名与端口号
                        //发送RPC请求
                        RpcResponse rpcResponse=send(remotingTransporter,serviceInfo.getHost(),serviceInfo.getPort());
                        //获取响应结果
                        if(rpcResponse==null){ 
                            log.error("send request failure",new IllegalStateException("response is null"));
                            return null;
                        }
                        if(rpcResponse.getException()!=null){ 
                            log.error("response has exception",rpcResponse.getException());
                            return null;
                        }
    
                        return rpcResponse.getResult();
                    }
            );
        }
    
        private RpcResponse send(RemotingTransporter remotingTransporter,String host,int port){ 
            log.info("send begin: "+host+":"+port);
            //客户端线程为1即可
            EventLoopGroup group=new NioEventLoopGroup(1);
            try { 
                //创建RPC连接
                Bootstrap bootstrap=new Bootstrap();
                bootstrap.group(group);
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() { 
                    @Override
                    protected void initChannel(SocketChannel channel) throws Exception { 
                        ChannelPipeline pipeline=channel.pipeline();
                        pipeline.addLast(new RemotingTransporterDecoder())
                                .addFirst(new RemotingTransporterEncoder())
                                .addLast(new RpcClientHandler(remotingTransporterMap));
                    }
                });
                ChannelFuture future=bootstrap.connect(host,port).sync();
                Channel channel=future.channel();
                log.info("invokeId: "+remotingTransporter.getInvokeId());
                //写入RPC请求对象
                channel.writeAndFlush(remotingTransporter).sync();
                channel.closeFuture().sync();
                log.info("send end");
                //获取RPC响应对象
                return (RpcResponse) remotingTransporterMap.get(remotingTransporter.getInvokeId()).getBodyContent();
            }catch (Exception e){ 
                log.error("client exception",e);
                return null;
            }finally { 
                group.shutdownGracefully();
                //移除请求编号和响应对象直接的映射关系
                remotingTransporterMap.remove(remotingTransporter.getInvokeId());
            }
    
        }
    
    }

[RPC]: https://blog.csdn.net/alinyua/column/info/29587
[https_github.com_linshenkx_rpc-netty-spring-boot-starter]: https://github.com/linshenkx/rpc-netty-spring-boot-starter
[Link 1]: https://blog.csdn.net/alinyua/article/details/84105650
[_ JDK_Hessian_Json_Protostuff_Xml_Avro_ProtocolBuffer_Thrift]: https://blog.csdn.net/alinyua/article/details/84201285
[RPC_ _ 2.0 _1]: https://blog.csdn.net/alinyua/article/details/84351255
[RPC_ _ 2.0 _2_RPC-Common]: https://blog.csdn.net/alinyua/article/details/84377228
[RPC_ _ 2.0 _3_RPC-Server_RPC-Client]: https://mp.csdn.net/mdeditor/84396060
[RPC_ _ 2.0 _4_BeanPostProcessor_RpcReference]: https://mp.csdn.net/mdeditor/84674301
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]: /images/20220414/08df125dbb8347a2bbc6b18db8345e00.png
[RPC-Client]: /images/20220414/ff697d620f864ac4882c403b50f7f8a9.png