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

爱被打了一巴掌 2022-04-18 04:12 212阅读 0赞

RPC-Client模块负责创建 动态代理对象 供 服务消费者 使用，而动态代理对象的方法执行则是通过RPC调用RPC-Server的服务实现。即RPC-Client屏蔽了底层的通信过程，使得服务消费者可以基于接口透明使用服务提供者的服务。

## 系列文章： ##

项目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 模块介绍
         *  3 工作流程
     *  二 pom文件
     *  三 简单组件：属性注入类和自动装配类
     *  四 ZKServiceDiscovery
     *  五 RpcClient
     *   *  create 方法
         *  send方法
         *  responseMap成员
     *  六 RpcClientHandler

## 一 介绍 ##

### 1 整体结构 ###

![整体结构][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]

### 2 模块介绍 ###

整体结构如下：

1.  RpcClient  
    RPC-Client模块核心类，也是服务消费者会直接使用到的类。该类负责根据传入的接口类，生成一个可以执行远程方法的动态代理对象。
2.  ZKServiceDiscovery  
    主要负责 **服务发现** ，根据服务名从ZK集群中获取相应RPC-Client地址信息
3.  RpcClientHandler  
    Rpc客户端处理器，将RPC-Server返回的RpcResponse对象存入管理。
4.  ZKProperties  
    属性注入类。
5.  RpcClientAutoConfiguration 和 spring.factories  
    封装成spring-boot-starter所需配置，开启自动装配。

### 3 工作流程 ###

服务消费者执行RpcClient的create方法

RpcClient利用ZKServiceDiscovery获取服务提供方信息

生成动态代理类

服务消费者调用动态代理类执行方法

建立Netty连接

将方法信息封装成RpcRequest发送

获取RpcResponse

提取结果返回

## 二 pom文件 ##

<?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-client-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>

## 三 简单组件：属性注入类和自动装配类 ##

属性装配类ZKProperties和RPC-Server的基本一致，自动装配类也是，原理相通。RpcClientAutoConfiguration源码如下

可以看到整个过程要管理的类其实比RPC-Server的要少很多。

/** * @version V1.0 * @author: lin_shen * @date: 2018/11/2 * @Description: TODO */
    @Configuration
    @EnableConfigurationProperties
    @ConditionalOnClass(RpcClient.class)
    public class RpcClientAutoConfiguration { 
        @ConditionalOnMissingBean
        @Bean
        public ZKProperties defaultZKProperties(){ 
            return new ZKProperties();
        }
    
        @ConditionalOnMissingBean
        @Bean
        public ZKServiceDiscovery zkServiceDiscovery(){ 
            return new ZKServiceDiscovery();
        }
    
        @Bean
        public RpcClient rpcClient(){ 
            return new RpcClient();
        }
    }

## 四 ZKServiceDiscovery ##

如下，核心方法为 discover，用于根据服务名从ZK集群获取到提供该服务的RPC-Server信息。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: zookeeper服务注册中心 */
    @Component
    @Log4j2
    @EnableConfigurationProperties(ZKProperties.class)
    public class ZKServiceDiscovery { 
    
      @Autowired
      private ZKProperties zkProperties;
    
      /** * 为客户端提供地址 * 根据服务名获取服务地址 * @param serviceName * @return */
      public String discover(String serviceName){ 
        ZkClient zkClient = new ZkClient(getAddress(zkProperties.getAddressList()), zkProperties.getSessionTimeOut(), zkProperties.getConnectTimeOut());
        try { 
          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);
          }
          //获取address节点地址
          String address=getRandomAddress(addressList);
          //获取address节点的值
          return zkClient.readData(servicePath+"/"+address);
        }finally { 
          zkClient.close();
        }
      }
      
      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()));
      }
      
    }

## 五 RpcClient ##

整个模块的核心类，用于生成服务的动态代理类并实现远程方法调用。

### create 方法 ###

根据传入接口类返回动态代理类  
主要使用了JDK的Proxy.newProxyInstance方法，其最关键的步骤就是实现InvocationHandler的invoke方法，这个时候便可以根据传入数据，构造封装成RpcRequest对象，然后利用 ZKServiceDiscovery 获取对于RPC-Server的地址信息。再调用send方法将RpcRequest发送到对应地址，并获取RpcResponse，再对RpcResult处理（如区分是否有异常等），返回最终的处理结果。完成invoke方法的实现。

### send方法 ###

根据RpcRequest对象和Rpc-Server地址信息，创建Netty连接，将RpcRequest发送到Rpc-Server并获取RpcResponse。

### responseMap成员 ###

responseMap的作用是作为一个中介暂存RpcResponse，因为Netty的RpcClientHandler在获取到RpcResponse对象还无法直接返回给send方法，所以就将其放入responseMap中，再由send方法根据RpcResponse对应的RequestId去获取。

需要注意，responseMap是 ConcurrentHashMap 类型的。因为目前的流程是每执行一次Rpc方法即执行一次Rpc调用都需要建立、断开netty连接，并且在这个过程需要根据requestId存放RpcResponse再将其移除。需要保证responseMap的线程安全。

/** * @version V1.0 * @author: lin_shen * @date: 2018/11/1 * @Description: TODO */
    @Log4j2
    @Component
    @AutoConfigureAfter(ZKServiceDiscovery.class)
    public class RpcClient { 
        @Autowired
        private ZKServiceDiscovery zkServiceDiscovery;
        /** * 存放请求编号与响应对象的映射关系 */
        private ConcurrentMap<String, RpcResponse> responseMap=new ConcurrentHashMap<>();
    
        @SuppressWarnings("unchecked")
        public <T> T create(final Class<?> interfaceClass){ 
            //创建动态代理对象
            return (T) Proxy.newProxyInstance(interfaceClass.getClassLoader(),
                    new Class<?>[]{ interfaceClass},
                    new InvocationHandler() { 
                        @Override
                        public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
                            //创建RPC请求对象
                            RpcRequest rpcRequest=new RpcRequest();
                            rpcRequest.setRequestId(UUID.randomUUID().toString());
                            rpcRequest.setInterfaceName(method.getDeclaringClass().getName());
                            rpcRequest.setMethodName(method.getName());
                            rpcRequest.setParameterTypes(method.getParameterTypes());
                            rpcRequest.setParameters(args);
                            //获取RPC服务地址
                            String serviceName=interfaceClass.getName();
                            String serviceAddress=zkServiceDiscovery.discover(serviceName);
                            log.info("get serviceAddres:"+serviceAddress);
                            //从RPC服务地址中解析主机名与端口号
                            String[] stringArray= StringUtils.split(serviceAddress,":");
                            String host= Objects.requireNonNull(stringArray)[0];
                            int port=Integer.parseInt(stringArray[1]);
                            //发送RPC请求
                            RpcResponse rpcResponse=send(rpcRequest,host,port);
                            //获取响应结果
                            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(RpcRequest rpcRequest,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 RpcEncoder(RpcRequest.class))
                                .addLast(new RpcDecoder(RpcResponse.class))
                                .addLast(new RpcClientHandler(responseMap));
                    }
                });
                ChannelFuture future=bootstrap.connect(host,port).sync();
                log.info("requestId: "+rpcRequest.getRequestId());
                //写入RPC请求对象
                Channel channel=future.channel();
                channel.writeAndFlush(rpcRequest).sync();
                channel.closeFuture().sync();
                log.info("send end");
                //获取RPC响应对象
                return responseMap.get(rpcRequest.getRequestId());
            }catch (Exception e){ 
                log.error("client exception",e);
                return null;
            }finally { 
                group.shutdownGracefully();
                //移除请求编号和响应对象直接的映射关系
                responseMap.remove(rpcRequest.getRequestId());
            }
    
        }
    
    }

## 六 RpcClientHandler ##

RpcClientHandler 只是用来接收数据，并不需要做什么处理，这里的作用就是将 RpcResponse 存入 responseMap 。

/** * @version V1.0 * @author: lin_shen * @date: 2018/11/1 * @Description: TODO */
    @Log4j2
    public class RpcClientHandler extends SimpleChannelInboundHandler<RpcResponse> { 
    
        private ConcurrentMap<String,RpcResponse> responseMap;
    
        public RpcClientHandler(ConcurrentMap<String,RpcResponse> responseMap){ 
            this.responseMap=responseMap;
        }
    
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, RpcResponse msg) throws Exception { 
            log.info("read a Response,requestId: "+msg.getRequestId());
            responseMap.put(msg.getRequestId(),msg);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception { 
            log.error("client caught exception",cause);
            ctx.close();
        }
    
    }

[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/da2c3df295014fb486b7abf3ded4d26d.png