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

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

系列文章：

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

手写通用类型负载均衡路由引擎（含随机、轮询、哈希等及其带权形式）
实现序列化引擎（支持 JDK默认、Hessian、Json、Protostuff、Xml、Avro、ProtocolBuffer、Thrift等序列化方式）
从零写分布式RPC框架系列 2.0 （1）架构升级
从零写分布式RPC框架系列 2.0 （2）RPC-Common模块设计实现
从零写分布式RPC框架系列 2.0 （3）RPC-Server和RPC-Client模块改造
从零写分布式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 结构图

结构图注意，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 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 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());
        }
    }
}