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

青旅半醒 2022-04-18 02:20 199阅读 0赞

RPC-Common模块提供RPC-Server和RPC-Client的通用对象，封装统一规则，使RPC Server和RPC Client 可以基于同一协议通信。主要包含底层通信的Netty所需的编码解码器（RpcEncoder，RpcDecoder），实现自定义协议的传输对象（RpcRequest、RpcResponse）以及编码解码器对Java对象序列化（反序列化）使用的工具 ProtoSerializationUtil。

## 系列文章： ##

项目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_]

### 文章目录 ###

*   *  系列文章：
     *  一 模块介绍
     *   *  结构
         *  流程图
     *  二 pom文件
     *  三 RpcRequest和RpcResponse
     *   *  RpcRequest
         *  RpcResponse
     *  四 序列化工具
     *  五 RpcEncoder和RpcDecoder
     *   *  RpcEncoder编码器
         *  RpcDecoder

## 一 模块介绍 ##

### 结构 ###

结构如下图  
![rpc-netty-common][]

### 流程图 ###

1.  RPC-Client封装服务请求发送到RPC-Server

封装请求信息

利用序列化工具编码

Netty通信

RPC-Client

RpcRequest

RpcEncoder

RPC-Server

1.  RPC-Server获取请求，执行业务处理，返回结果

利用反序列化工具解码

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

获取结果

利用序列化工具编码

Netty通信

RPC-Server

RpcDecoder

RpcServerHandler

RpcResponse

RpcEncoder

RPC-Client

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

利用反序列化工具解码

RPC-Client

RpcDecoder

Rpcponse

获得最终执行结果

## 二 pom文件 ##

RpcRequest和RpcResponse是Java bean，不需要多余的依赖。  
RpcEncoder和RpcDecoder是Netty编解码器，依赖于netty-all。  
ProtoSerializationUtil是基于ProtoStuff的序列化工具，并依赖objenesis实现更高级的反射功能，进行对象反序列化。

<?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-common</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.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <optional>true</optional>
            </dependency>
            <dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
            </dependency>
            <dependency>
                <groupId>io.protostuff</groupId>
                <artifactId>protostuff-core</artifactId>
            </dependency>
            <dependency>
                <groupId>io.protostuff</groupId>
                <artifactId>protostuff-runtime</artifactId>
            </dependency>
            <dependency>
                <groupId>org.objenesis</groupId>
                <artifactId>objenesis</artifactId>
            </dependency>
        </dependencies>
        
    </project>

## 三 RpcRequest和RpcResponse ##

### RpcRequest ###

RpcRequest是Rpc请求对象，其属性相当于自定义协议规则，如下  
其中requestId是为了区分对相同服务的不同请求，还可以加上超时时间、版本信息等来定制自己的协议

@Data
    public class RpcRequest { 
        /** * 请求ID */
        private String requestId;
        /** * 接口名称 */
        private String interfaceName;
        /** * 方法名 */
        private String methodName;
        /** * 方法参数类型列表 */
        private Class<?>[] parameterTypes;
        /** * 参数列表 */
        private Object[] parameters;
    }

### RpcResponse ###

以下为最简单也是最基本的RpcResponse，复杂一点可以区分消息头和消息体，在消息头里指定编码长度，编码类型等

@Data
    public class RpcResponse { 
        /** * 对应请求的requestId */
        private String requestId;
        /** * 异常信息 */
        private Exception exception;
        /** * 响应结果 */
        private Object result;
    }

## 四 序列化工具 ##

目前只提供了基于protostuff的序列化工具，后面可以升级成一个序列化引擎，可以动态选择序列化方式  
protostuff的GitHub地址为：[https://github.com/protostuff/protostuff][https_github.com_protostuff_protostuff]

public class ProtoSerializationUtil { 
    
        private static final Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();
    
        private static final Objenesis objenesis = new ObjenesisStd(true);
    
        /** * 序列化（对象 -> 字节数组） */
        @SuppressWarnings("unchecked")
        public static <T> byte[] serialize(T obj) { 
            Class<T> cls = (Class<T>) obj.getClass();
            LinkedBuffer buffer = LinkedBuffer.allocate(LinkedBuffer.DEFAULT_BUFFER_SIZE);
            try { 
                Schema<T> schema = getSchema(cls);
                return ProtostuffIOUtil.toByteArray(obj, schema, buffer);
            } catch (Exception e) { 
                throw new IllegalStateException(e.getMessage(), e);
            } finally { 
                buffer.clear();
            }
        }
    
        /** * 反序列化（字节数组 -> 对象） */
        public static <T> T deserialize(byte[] data, Class<T> cls) { 
            try { 
                T message = objenesis.newInstance(cls);
                Schema<T> schema = getSchema(cls);
                ProtostuffIOUtil.mergeFrom(data, message, schema);
                return message;
            } catch (Exception e) { 
                throw new IllegalStateException(e.getMessage(), e);
            }
        }
    
        @SuppressWarnings("unchecked")
        private static <T> Schema<T> getSchema(Class<T> cls) { 
            Schema<T> schema = (Schema<T>) cachedSchema.get(cls);
            if (schema == null) { 
                schema = RuntimeSchema.createFrom(cls);
                cachedSchema.put(cls, schema);
            }
            return schema;
        }
    }

## 五 RpcEncoder和RpcDecoder ##

### RpcEncoder编码器 ###

编码器负责将Java对象序列化成字节数组，再封装在Netty 的ByteBuf里面

public class RpcEncoder extends MessageToByteEncoder { 
    
        private Class<?> genericClass;
    
        public RpcEncoder(Class<?> genericClass){ 
            this.genericClass=genericClass;
        }
    
        @Override
        protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { 
            if(genericClass.isInstance(msg)){ 
                byte[] data= ProtoSerializationUtil.serialize(msg);
                out.writeInt(data.length);
                out.writeBytes(data);
            }
        }
    }

### RpcDecoder ###

发送的时候比较简单，因为byteBuf是动态扩展的。但接收的时候就需要考虑半包、粘包的问题了。这里只是用了(in.readableBytes()<4)来简单处理，但如果传输的对象比较大，就应该考虑加其他的Decoder来解决。这里暂未扩展。

public class RpcDecoder extends ByteToMessageDecoder { 
        private Class<?> genericClass;
    
        public RpcDecoder(Class<?> genericClass){ 
            this.genericClass=genericClass;
        }
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
            if(in.readableBytes()<4){ 
                return;
            }
            in.markReaderIndex();
            int dataLength=in.readInt();
            if(in.readableBytes()<dataLength){ 
                in.resetReaderIndex();
                return;
            }
            byte[] data=new byte[dataLength];
            in.readBytes(data);
            out.add(ProtoSerializationUtil.deserialize(data,genericClass));
        }
    }

[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
[rpc-netty-common]: /images/20220418/884905991674409ab3e3dbfbf84e3fe9.png
[https_github.com_protostuff_protostuff]: https://github.com/protostuff/protostuff