手写RPC框架（十三）

v2.4

更新：增加序列化的实现hessian,json，Thrift, FST、Avro等序列化工具，从这个版本开始JDK自带的序列化和protoc编译后的进行序列化的暂时先不使用了

进行序列化工具的提取

package serialization;

import annotation.CodecSelector;
import exception.RpcException;
import serialization.fst.FSTUtils;
import serialization.hessian.HessianUtils;
import serialization.kryo.KryoUtils;
import serialization.protostuff.ProtostuffUtils;

//这是一个进行统一序列化的一个工具
public class SerializationTool implements Serializer {

    static String codec = Serialization.class.getAnnotation(CodecSelector.class).Codec();

    @Override
    public byte[] serialize(Object obj) throws RpcException {
        switch (codec) {
            case "kryo":
                return new KryoUtils().serialize(obj);
            case "protostuff":
                return new ProtostuffUtils().serialize(obj);
            case "hessian":
                return new HessianUtils().serialize(obj);
            case "fst":
                return new FSTUtils().serialize(obj);
            case "thrift":
                return null;
            case "avro":
                return null;
            case "jackson":
                return null;
            case "fastjson":
                return null;
            case "gson":
                return null;
            default:
                throw new RpcException("你所找的序列化方法还没编写，或者可能在该版本被废弃了，你可以看看2.2版本");
        }
    }

    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException {
        switch (codec) {
            case "kryo":
                return new KryoUtils().deserialize(bytes, clazz);
            case "protostuff":
                return new ProtostuffUtils().deserialize(bytes, clazz);
            case "hessian":
                return new HessianUtils().deserialize(bytes, clazz);
            case "fst":
                return new FSTUtils().deserialize(bytes, clazz);
            case "thrift":
                return null;
            case "avro":
                return null;
            case "jackson":
                return null;
            case "fastjson":
                return null;
            case "gson":
                return null;
            default:
                throw new RpcException("你所找的序列化方法还没编写，或者可能在该版本被废弃了，你可以看看2.2版本");
        }
    }
}

handler中的改造（顺便改造了下clientHandler 因为之前改写了以后，string也要序列化，然后返回类型的判断按照之前出了问题，所以传入了方法，获取返回的类型）

package consumer.netty_client_handler;


import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import serialization.SerializationTool;

import java.lang.reflect.Method;
import java.util.concurrent.Callable;



//实现了Callable接口实现了异步调用

public class NettyClientHandler24 extends ChannelInboundHandlerAdapter implements Callable{
    //传入的参数
    private Object param;
    private Method method;
    private Object response;
    private ChannelHandlerContext context;

    //序列化工具
    static SerializationTool serializationTool = new SerializationTool();

    public void setParam(Object param) {
        this.param = param;
    }
    public void setMethod(Method method) {
        this.method = method;
    }

    //当成功建立 就赋值上下文对象
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        context = ctx;
        System.out.println("U•ェ•*U 成功连接");
    }

    @Override
    public synchronized void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {

        //在这要进行解码 获得传回来的信息  如果是遇到下面的msg 那就代表传回来的肯定是个byte[]
        // 根据我们要的方法进行解码 传回来的应该是方法的response的
        //根据需要的返回类型进行反序列化
        msg = serializationTool.deserialize((byte[]) msg,method.getReturnType());

        response = msg;
        notify();
    }

    //调用的时候  就进行传输
    @Override
    public synchronized Object call() throws Exception {

        //这个变量的目的就是保留原来的param实际参数类型，当返回的时候 可以当作反序列化的模板
        Object request = param;
        //判断是否需要protostuff进行序列化 因为使用这个进行序列话 是我没有相应的解码器 2.4之后 就算是string也进行序列化

        request = serializationTool.serialize(request);

        context.writeAndFlush(request);
        wait();
        return response;
    }

    //异常处理
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

实现hessian

最简单一步引入依赖



    com.caucho
    hessian
    4.0.65

实现序列化反序列化的功能

package serialization.hessian;

import com.caucho.hessian.io.HessianInput;
import com.caucho.hessian.io.HessianOutput;
import exception.RpcException;
import serialization.Serializer;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;


public class HessianUtils implements Serializer {

    //进行序列化
    @Override
    public byte[] serialize(Object obj) throws RpcException {
        try {
            ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
            HessianOutput hessianOutput = new HessianOutput(outputStream);
            hessianOutput.writeObject(obj);
            return outputStream.toByteArray();
        } catch (IOException e) {
            e.printStackTrace();
            throw new RpcException("序列化失败");
        }
    }

    //进行反序列化
    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException {
        ByteArrayInputStream inputStream = new ByteArrayInputStream(bytes);
        HessianInput hessianInput = new HessianInput(inputStream);
        try {
            Object object = hessianInput.readObject(clazz);
            return clazz.cast(object);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RpcException("反序列化失败");
        }
    }
}

实现json序列化

首先Json序列化有一系列的框架实现（fastJson、gson、Jackson）

Jackson

依赖引用

  

    com.fasterxml.jackson.core
    jackson-databind
    2.9.5

实现序列化反序列化功能

package serialization.json;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import exception.RpcException;
import serialization.Serializer;

import java.io.IOException;

public class JacksonUtils implements Serializer {

    //Jackson中的关键
    static ObjectMapper mapper = new ObjectMapper();

    @Override
    public byte[] serialize(Object obj) throws RpcException, JsonProcessingException {
        //将obj转换成json再将json转换成字节数组
        return mapper.writeValueAsBytes(obj);
    }

    //将json字符串或者字节组转换为对应的类
    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException, IOException {
        return mapper.readValue(bytes,clazz);
    }
}

fastJson

依赖引用



    com.alibaba
    fastjson
    1.2.80

实现序列化反序列化功能

package serialization.json;

import com.alibaba.fastjson.JSON;
import exception.RpcException;
import serialization.Serializer;
import java.nio.charset.StandardCharsets;

//阿里巴巴旗下的进行json转换的工具
public class FastJsonUtils implements Serializer {
    @Override
    public byte[] serialize(Object obj) throws RpcException {

        // return JSON.toJSONString(obj,true).getBytes(StandardCharsets.UTF_8);
        return JSON.toJSONBytes(obj);
    }

    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException {
        //后面的序列化参数也是可选可不选
        return JSON.parseObject(bytes,clazz);
    }
}

问题：序列化后对象中的属性变为null
解决：需要对对象中的属性设置get set 方法

gson

依赖引用



    com.google.code.gson
    gson
    2.8.8

实现序列化反序列化功能

package serialization.json;

import com.google.gson.Gson;
import exception.RpcException;
import serialization.Serializer;

import java.nio.charset.StandardCharsets;

//gson
public class GsonUtils implements Serializer {
    static Gson gson = new Gson();
    @Override
    public byte[] serialize(Object obj) throws RpcException {
        return gson.toJson(obj).getBytes(StandardCharsets.UTF_8);
    }
    
	//重点就是上下的编解码器  一定要保证字符串的正确性
    
    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException {
        return gson.fromJson(new String(bytes,StandardCharsets.UTF_8),clazz);
    }
}

问题：com.google.gson.JsonSyntaxException: java.lang.IllegalStateException: Expected BEGIN_OBJECT but was BEGIN_ARRAY at line 1 column 2 path $
解决：保证传输的字符串编解码正确

实现FST序列化工具

引入依赖 (发现fst和nacos的冲突2.57就不行会报错)
```
    de.ruedigermoeller
    fst
    2.04
```

工具类的实现

package serialization.fst;

import exception.RpcException;
import serialization.Serializer;
import org.nustaq.serialization.FSTConfiguration;

//一款较新的序列化工具
public class FSTUtils implements Serializer {

    //需要弄懂 rather than
    static FSTConfiguration fstConfiguration = FSTConfiguration.createStructConfiguration();

    //进行序列化
    @Override
    public byte[] serialize(Object obj) throws RpcException {
        return fstConfiguration.asByteArray(obj);
    }

    //进行反序列化
    @Override
    public  T deserialize(byte[] bytes, Class clazz) throws RpcException {
        return clazz.cast(fstConfiguration.asObject(bytes));
    }
}

手写RPC框架（十三）

Java相关栏目本月热门文章