从零写分布式RPC框架系列 2.0 （2）RPC-Common模块设计实现

怼烎@ 2022-04-14 02:18 387阅读 0赞

RPC-Common模块相对于1.0版本复杂了很多，最主要的变化在于将 Rpc的Netty处理器从RPC-Server和RPC-Client收回。1.0 版本的设计思路是尽可能减少冗余依赖，所以RPC-Common一般只放通用的功能。现在则是尽可能都放在RPC-Common模块，以方便工程升级复杂化后的统一协调管理。以后功能将集中在一个模块下（名字不一定还是RPC-Common），RPC-Server和RPC-Client则会担任将功能选择性开放给使用者的轻量级角色。部分内容在 [从零写分布式RPC框架 系列 1.0 （2）RPC-Common模块设计实现][RPC_ _ 1.0 _2_RPC-Common] 已有说明，本文不再赘述。

## 系列文章： ##

专栏：[从零开始写分布式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]

### 文章目录 ###

*   *  系列文章：
     *  一 模块介绍
     *   *  结构图
         *  流程图
     *  二 协议相关实体类
     *   *  RemotingTransporter
         *  BodyContent及其实现类
     *  三 Netty编码器和解码器
     *   *  RemotingTransporterEncoder 编码器
         *  RemotingTransporterDecoder 解码器
     *  四 Netty服务端和客户端处理器
     *   *  RpcServerHandler 服务端处理器
         *  RpcClientHandler 客户端处理器

## 一 模块介绍 ##

### 结构图 ###

![结构图][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FsaW55dWE_size_16_color_FFFFFF_t_70]

### 流程图 ###

1.  RPC-Client 将服务请求封装到远程传输实体里，经过编码发送到RPC-Server
    
    封装服务调用信息
    
    根据协议类型和序列化方式等进一步封装
    
    Netty通信
    
    RPC-Client
    
    RpcRequest
    
    RemotingTransporter
    
    RPC-Server
2.  RPC-Server获取请求，执行业务处理，返回结果

根据RemotingTransporter消息头利用反序列化工具解码获取请求信息

利用请求信息反射执行方法执行业务逻辑

获取结果

根据协议类型和序列化方式等进一步封装

Netty通信

RPC-Server

RpcRequest

RpcServerHandler

RpcResponse

RemotingTransporter

RPC-Client

1.  RPC-Client获取结果进行处理

根据协议类型和序列化方式解码

RPC-Client

RemotingTransporter

Rpcponse

获得最终执行结果

## 二 协议相关实体类 ##

### RemotingTransporter ###

这里需要注意的是invokeId我使用的是原子递增的方式来确保唯一性，因为只需要对单客户端保证唯一性，所以这种方式应该是够用的。如果单客户端有高并发请求的话（虽然不太合理），可以考虑其他方式。

Flag只占一个字节大小，但我这里实现实现的效率比较低，后面应该得优化。

另外消息体BodyContent只是个接口，根据消息头的标识，BodyContent可以代表不同的内容。

/** * @version V1.0 * @author: lin_shen * @date: 18-11-12 * @Description: 自定义远程传输实体 (magic+flag+invokeId+bodyLength+bodyContent) */
    @Data
    @Builder
    @AllArgsConstructor
    @NoArgsConstructor
    public class RemotingTransporter { 
    
        /** * 魔数 */
        public static final short MAGIC=(short)0x9826;
    
        /** * 用原子递增的方式来获取不重复invokeId */
        private static final AtomicLong invokeIdGnerator=new AtomicLong(0L);
    
        /** * 标志位, 一共8个地址位。 * 低四位用来表示消息体数据用的序列化工具的类型 * 高四位中，第一位为1表示是request请求，为0表示是reponse应答 * TODO:第二位为1表示双向传输（即有返回response） * TODO:第三位为1表示是心跳ping事件 * TODO:预留位 */
        private Flag flag;
    
        @Getter
        @ToString
        public static class Flag{ 
            private boolean isRequest;
            private boolean isTwoway;
            private boolean isPing;
            private boolean isOther;
    
            private int serializeType;
    
            private byte thisByte;
    
            public Flag(boolean isRequest, boolean isTwoway, boolean isPing, boolean isOther, int serializeType) { 
    
                if(serializeType<0||serializeType>15){ 
                    throw new IllegalArgumentException("serializeType 对应整数应该在 0 到 15 之间");
                }
    
                this.isRequest = isRequest;
                this.isTwoway = isTwoway;
                this.isPing = isPing;
                this.isOther = isOther;
                this.serializeType = serializeType;
    
                int byteTem= (isRequest?1:0)<<7;
                byteTem=byteTem | ((isTwoway?1:0)<<6);
                byteTem=byteTem | ((isPing?1:0)<<5);
                byteTem=byteTem | ((isOther?1:0)<<4);
                byteTem=byteTem | serializeType;
    
                this.thisByte= (byte) byteTem;
            }
    
            public Flag(byte thisByte){ 
                this.thisByte=thisByte;
    
                isRequest=((thisByte>>>7)&1)==1;
                isTwoway=((thisByte>>>6)&1)==1;
                isPing=((thisByte>>>5)&1)==1;
                isOther=((thisByte>>>4)&1)==1;
    
                serializeType=thisByte & 15;
    
            }
    
        }
    
        /** * 每一个请求的唯一识别id（由于采用异步通讯的方式，用来把请求request和返回的response对应上） */
        private long invokeId=invokeIdGnerator.getAndIncrement();
    
        /** * 消息体字节数组长度 */
        private int bodyLength;
    
        /** * 消息体内容(还需要编码序列化成字节数组) */
        private transient BodyContent bodyContent;
    
    }

### BodyContent及其实现类 ###

BodyContent只起标识作用，本身无规定任何方法。在传输过程中以字节数组形式存在，由 RemotingTransporter 的消息头 标识其代表类型及序列化方式。

目前其实现类有 RpcRequest 和 RpcResponse，分别代表Rpc请求和Rpc响应。注意这些实现类应该只包含自身业务信息，其他的应由 RemotingTransporter 统一标识处理。

## 三 Netty编码器和解码器 ##

### RemotingTransporterEncoder 编码器 ###

编码器的实现比较简单。  
这里需要注意，相对于1.0版本 编码器的处理类型由构造方法动态传入的设计，这里直接限制了处理类型为 RemotingTransporter 。这是因为在1.0版本中，不管是RpcRequest还是RpcResponse都会被编码，而我们并不想对每种类型都专门写一个编码器，所以一个比较讨巧的做法是将类型动态传入。  
而在2.0版本中，因为我们统一了不管是 RpcRequest还是 RpcResponse在传输过程中其表现实体都为RemotingTransporter，将其类型信息封装到消息头里，屏蔽了各自传输实体的差异，所以直接指定 RemotingTransporter 即可。而且还可以减少类型信息和代码的高耦合，将不同类型不同编解码方式的判断和处理收归到编解码器里。

/** * @version V1.0 * @author: lin_shen * @date: 18-11-12 * @Description: TODO */
    @Log4j2
    public class RemotingTransporterEncoder extends MessageToByteEncoder<RemotingTransporter> { 
    
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, RemotingTransporter remotingTransporter, ByteBuf byteBuf) throws Exception { 
            //获取请求体数组
            //使用序列化引擎
            byte[] body= SerializerEngine.serialize(remotingTransporter.getBodyContent(), SerializeTypeEnum.queryByCode(remotingTransporter.getFlag().getSerializeType()));
            //magic+flag+invokeId+bodyLength+bodyContent
            byteBuf.writeShort(RemotingTransporter.MAGIC)
                    .writeByte(remotingTransporter.getFlag().getThisByte())
                    .writeLong(remotingTransporter.getInvokeId())
                    .writeInt(body.length)
                    .writeBytes(body);
            log.info("write end");
    
        }
    
    }

### RemotingTransporterDecoder 解码器 ###

需要注意，这里解码器没有继承自ByteToMessageDecoder而是使用了ReplayingDecoder（当然它也继承自ByteToMessageDecoder），其泛型参数指定了用于状态管理的类型（不需要状态管理可以使用Void），这里我们是用了内部枚举类 RemotingTransporterDecoder.State。

其工作原理可简单理解为：ByteBuf源源不断地传递给解码器，解码器需要根据当前状态判断应该执行什么读取操作以进行解码。因此一个状态通常代表着传输体的一个变量，根据前面所说，这里的状态应该有：**HEADER\_MAGIC, HEADER\_FLAG, HEADER\_INVOKE\_ID, HEADER\_BODY\_LENGTH, BODY** 。另外需要注意读取是状态时连续且循环的，因此在根据状态 switch 时应按顺序编写，并可不使用 break 以提高效率。每个状态对应的解码操作执行完的时候都应该将状态及时切换，最后一个状态执行完切换回默认初始状态。最初的初始状态在父类构造方法传入。

/** * @version V1.0 * @author: lin_shen * @date: 18-11-12 * @Description: TODO */
    @Log4j2
    public class RemotingTransporterDecoder extends ReplayingDecoder<RemotingTransporterDecoder.State> { 
    
        private static final int MAX_BODY_SIZE = 1024 * 1024 * 5;
    
        /** * 用于暂存解码RemotingTransporter信息,一个就够了 */
        private static RemotingTransporter remotingTransporter=RemotingTransporter.builder().build();
    
        /** * 用于ReplayingDecoder的状态管理 */
        enum State { 
            HEADER_MAGIC, HEADER_FLAG, HEADER_INVOKE_ID, HEADER_BODY_LENGTH, BODY
        }
    
        public RemotingTransporterDecoder( ){ 
            //设置 state() 的初始值,以便进入switch
            super(State.HEADER_MAGIC);
        }
    
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception { 
            //注意这里在BODY之前都没有break
            switch (this.state()){ 
                case HEADER_MAGIC:
                    checkMagic(byteBuf.readShort());
                    //移到下一检查点(一是改变state的值的状态，二是获取到最新的读指针的下标)
                    checkpoint(State.HEADER_FLAG);
                case HEADER_FLAG:
                    remotingTransporter.setFlag(new RemotingTransporter.Flag(byteBuf.readByte()));
                    checkpoint(State.HEADER_INVOKE_ID);
                case HEADER_INVOKE_ID:
                    remotingTransporter.setInvokeId(byteBuf.readLong());
                    checkpoint(State.HEADER_BODY_LENGTH);
                case HEADER_BODY_LENGTH:
                    remotingTransporter.setBodyLength(byteBuf.readInt());
                    checkpoint(State.HEADER_BODY_LENGTH);
                case BODY:
                    int bodyLength = checkBodyLength(remotingTransporter.getBodyLength());
                    byte[] bytes=new byte[bodyLength];
                    byteBuf.readBytes(bytes);
                    Class genericClass=remotingTransporter.getFlag().isRequest()?RpcRequest.class: RpcResponse.class;
                    BodyContent bodyContent= (BodyContent) SerializerEngine.deserialize(bytes,genericClass,SerializeTypeEnum.queryByCode(remotingTransporter.getFlag().getSerializeType()));
                    RemotingTransporter remotingTransporter1=RemotingTransporter.builder()
                            .flag(remotingTransporter.getFlag())
                            .invokeId(remotingTransporter.getInvokeId())
                            .bodyLength(remotingTransporter.getBodyLength())
                            .bodyContent(bodyContent)
                            .build();
                    list.add(remotingTransporter1);
                    break;
                default:
                    break;
            }
            //顺利读完body后应置回起点
            checkpoint(State.HEADER_MAGIC);
    
        }
    
        private int checkBodyLength(int bodyLength) throws RemotingContextException { 
            if (bodyLength > MAX_BODY_SIZE) { 
                throw new RemotingContextException("body of request is bigger than limit value "+ MAX_BODY_SIZE);
            }
            return bodyLength;
        }
    
        private void checkMagic(short magic) throws RemotingContextException{ 
            //检查魔数
            if (RemotingTransporter.MAGIC != magic) { 
                log.error("魔数不匹配");
                throw new RemotingContextException("magic value is not equal "+RemotingTransporter.MAGIC);
            }
        }
    
    }

## 四 Netty服务端和客户端处理器 ##

### RpcServerHandler 服务端处理器 ###

服务端处理器的主要变化时增加了Map<String, Semaphore> serviceSemaphoreMap用于限制不同服务的工作线程数。使其同一时间进入handle的线程数受到限制。

/** * @version V1.0 * @author: lin_shen * @date: 2018/10/31 * @Description: RPC服务端处理器（处理RpcRequest） */
    @Log4j2
    public class RpcServerHandler extends SimpleChannelInboundHandler<RemotingTransporter> { 
    
      /** * 存放 服务名称 与 服务实例 之间的映射关系 */
      private final Map<String, Object> handlerMap;
    
      /** * 存放 服务名称 与 信号量 之间的映射关系 * 用于限制每个服务的工作线程数 */
      private final Map<String, Semaphore> serviceSemaphoreMap;
    
      public RpcServerHandler(Map<String, Object> handlerMap,Map<String, Semaphore> serviceSemaphoreMap) { 
        this.handlerMap = handlerMap;
        this.serviceSemaphoreMap=serviceSemaphoreMap;
      }
    
      @Override
      protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingTransporter remotingTransporter) throws Exception { 
        log.info("channelRead0 begin");
        remotingTransporter.setFlag(new RemotingTransporter.Flag(false,true,false,false,remotingTransporter.getFlag().getSerializeType()));
        RpcResponse rpcResponse=new RpcResponse();
        RpcRequest rpcRequest=(RpcRequest)remotingTransporter.getBodyContent();
        Semaphore semaphore = serviceSemaphoreMap.get(rpcRequest.getInterfaceName());
        boolean acquire=false;
            try { 
            // 处理 RPC 请求成功
            log.info("进入限流");
            acquire=semaphore.tryAcquire();
            if(acquire){ 
              Object result= handle(rpcRequest);
              rpcResponse.setResult(result);
            }
    
          } catch (Exception e) { 
            // 处理 RPC 请求失败
            rpcResponse.setException(e);
            log.error("handle result failure", e);
          } finally { 
            if(acquire){ 
              semaphore.release();
              log.info("释放信号量");
            }
          }
        remotingTransporter.setBodyContent(rpcResponse);
        channelHandlerContext.writeAndFlush(remotingTransporter).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");
        // 获取服务实例
        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);
        return method.invoke(serviceBean, parameters);
      }
    
    
    }

### RpcClientHandler 客户端处理器 ###

主要是将请求Id对应Response改成了对应RemotingTransporter，其他基本不变

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

[RPC_ _ 1.0 _2_RPC-Common]: https://blog.csdn.net/alinyua/article/details/83819288
[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/1d9570c135a14ceb95b93f4a4c0ab2d8.png