Kryo+Netty传输序列化对象

此处观看更加
本文的完整代码可以在此查看

这段时间不是一直想要使用Netty模仿一下dubbo，自己写一个RPC框架嘛，然后在学习的过程中还是学到了不少新东西。我们知道在网络传输中，数据都是以二进制流来传输，但是在Java中数据都是以对象的形式来存储，所以我们想要传输数据，这就涉及到对象的序列化以及反序列化了。而我们知道，不同的序列化协议适用不同的应用场景，jdk原生的序列化方式因为其性能原因绝大多数的人都不会考虑使用它，而我们想要写一个高性能的RPC框架，一个合适的序列化协议自然也是重中之重，因为目前所构思的RPC框架只是 Java to Java,所以我选择Kryo作为序列/反序列化的方式。

其他的序列化方式还有很多，他们都各自不同的优缺点，和不同的使用场景。想要深入了解的同学不妨参考一下下面的的文章：

序列化协议是应用层的协议

kryo的使用方法可以参靠下面这篇文章：

目录结构
C:.
├─client 客户端代码
├─codec 编解码器
├─entity 实体类
└─server 服务端代码

首先，先导入Netty的依赖，我是直接导入的Netty的所有模块

        
            io.netty
            >netty-all
            4.1.68.Final
        

然后导入Kryo的相关依赖，因为我使用的KryoUtil，还需要导入commons-codec的依赖,因为spring的相关依赖会和Kryo的依赖冲突，所以直接导入的Kryo-shaded，具体的原因读者可以自行百度，我不再赘述

        
            com.esotericsoftware
            kryo-shaded
            4.0.0
        
        
            commons-codec
            commons-codec
            1.10
        

再导入一些其他的依赖

 
        
            org.junit.jupiter
            junit-jupiter-api
            5.8.2
            test
        
        
        
            org.apache.logging.log4j
            log4j-slf4j-impl
            2.6.2
        
        
            org.apache.logging.log4j
            log4j-core
            2.6.2
        
        
        
            org.projectlombok
            lombok
            1.16.16
        

编写Kryo工具类，用于后面的序列反序列化对象

package cuit.epoch.pymjl.util;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import org.apache.commons.codec.binary.Base64;
import org.objenesis.strategy.StdInstantiatorStrategy;

import java.io.*;

public class KryoUtil {
    private static final String DEFAULT_ENCODING = "UTF-8";

    //每个线程的 Kryo 实例
    private static final ThreadLocal kryoLocal = new ThreadLocal() {
        @Override
        protected Kryo initialValue() {
            Kryo kryo = new Kryo();

            
            //支持对象循环引用（否则会栈溢出）
            kryo.setReferences(true); //默认值就是 true，添加此行的目的是为了提醒维护者，不要改变这个配置

            //不强制要求注册类（注册行为无法保证多个 JVM 内同一个类的注册编号相同；而且业务系统中大量的 Class 也难以一一注册）
            kryo.setRegistrationRequired(false); //默认值就是 false，添加此行的目的是为了提醒维护者，不要改变这个配置

            //Fix the NPE bug when deserializing Collections.
            ((Kryo.DefaultInstantiatorStrategy) kryo.getInstantiatorStrategy())
                    .setFallbackInstantiatorStrategy(new StdInstantiatorStrategy());

            return kryo;
        }
    };

    
    public static Kryo getInstance() {
        return kryoLocal.get();
    }

    //-----------------------------------------------
    //          序列化/反序列化对象，及类型信息
    //          序列化的结果里，包含类型的信息
    //          反序列化时不再需要提供类型
    //-----------------------------------------------

    
    public static  byte[] writeToByteArray(T obj) {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);

        Kryo kryo = getInstance();
        kryo.writeClassAndObject(output, obj);
        output.flush();

        return byteArrayOutputStream.toByteArray();
    }

    
    public static  String writeToString(T obj) {
        try {
            return new String(Base64.encodeBase64(writeToByteArray(obj)), DEFAULT_ENCODING);
        } catch (UnsupportedEncodingException e) {
            throw new IllegalStateException(e);
        }
    }

    
    @SuppressWarnings("unchecked")
    public static  T readFromByteArray(byte[] byteArray) {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArray);
        Input input = new Input(byteArrayInputStream);

        Kryo kryo = getInstance();
        return (T) kryo.readClassAndObject(input);
    }

    
    public static  T readFromString(String str) {
        try {
            return readFromByteArray(Base64.decodeBase64(str.getBytes(DEFAULT_ENCODING)));
        } catch (UnsupportedEncodingException e) {
            throw new IllegalStateException(e);
        }
    }

    //-----------------------------------------------
    //          只序列化/反序列化对象
    //          序列化的结果里，不包含类型的信息
    //-----------------------------------------------

    
    public static  byte[] writeObjectToByteArray(T obj) {
        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
        Output output = new Output(byteArrayOutputStream);

        Kryo kryo = getInstance();
        kryo.writeObject(output, obj);
        output.flush();

        return byteArrayOutputStream.toByteArray();
    }

    
    public static  String writeObjectToString(T obj) {
        try {
            return new String(Base64.encodeBase64(writeObjectToByteArray(obj)), DEFAULT_ENCODING);
        } catch (UnsupportedEncodingException e) {
            throw new IllegalStateException(e);
        }
    }

    
    @SuppressWarnings("unchecked")
    public static  T readObjectFromByteArray(byte[] byteArray, Class clazz) {
        ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(byteArray);
        Input input = new Input(byteArrayInputStream);

        Kryo kryo = getInstance();
        return kryo.readObject(input, clazz);
    }

    
    public static  T readObjectFromString(String str, Class clazz) {
        try {
            return readObjectFromByteArray(Base64.decodeBase64(str.getBytes(DEFAULT_ENCODING)), clazz);
        } catch (UnsupportedEncodingException e) {
            throw new IllegalStateException(e);
        }
    }

}

1. 编写RPC请求类

package cuit.epoch.pymjl.nettydemo.entity;

import lombok.*;


@AllArgsConstructor
@NoArgsConstructor
@Data
@Builder
@ToString
public class RpcRequest {
    private String interfaceName;
    private String methodName;
}

2. 编写RPC响应类

package cuit.epoch.pymjl.nettydemo.entity;

import lombok.*;


@NoArgsConstructor
@AllArgsConstructor
@Data
@ToString
@Builder
public class RpcResponse {
    private String message;
}

注意，对数据进行编码时因为TCP粘包/拆包的原因，我们这里需要自定义传输协议，然后我这里是：把传输数据的长度写在字节数组的前面四个字节中，服务端在读取数据时会先从前四个字节获取到这次传输数据的长度，在对数据进行都写操作

另外，我们需要在编解码器中将对象序列化成字节数组或者将字节数组反序列化成原对象

1. 编码器

package cuit.epoch.pymjl.nettydemo.codec;

import cuit.epoch.pymjl.util.KryoUtil;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.AllArgsConstructor;


@AllArgsConstructor
public class NettyKryoEncoder extends MessageToByteEncoder