Netty（六）之protostuff

CBeann

发布于 2023-12-25 17:23:26

1380

发布于 2023-12-25 17:23:26

文章被收录于专栏：CBeann的博客CBeann的博客

protostuff和protobuf的区别

protostuff是一个基于protobuf实现的序列化方法
在几乎不损耗性能的情况下做到了不用我们写.proto文件来实现序列化
使用它也非常简单，所以直接上代码。

操作流程

前提

在

Netty（一）之helloworld Netty（一）之helloworld_CBeann的博客-CSDN博客

的基础之上修改

pom

<dependency>
            <groupId>org.objenesis</groupId>
            <artifactId>objenesis</artifactId>
            <version>2.6</version>
        </dependency>

        <dependency>
            <groupId>com.dyuproject.protostuff</groupId>
            <artifactId>protostuff-api</artifactId>
            <version>1.0.10</version>
        </dependency>
        <dependency>
            <groupId>com.dyuproject.protostuff</groupId>
            <artifactId>protostuff-core</artifactId>
            <version>1.0.10</version>
        </dependency>
        <dependency>
            <groupId>com.dyuproject.protostuff</groupId>
            <artifactId>protostuff-runtime</artifactId>
            <version>1.0.10</version>
        </dependency>
        <dependency>
            <groupId>io.netty</groupId>
            <artifactId>netty-all</artifactId>
            <version>5.0.0.Alpha1</version>
        </dependency>

创建自定义的POJO

public class MsgInfo {

    private Integer id;
    private String msg;

//getter setter  构造方法省略。。。
}

创建序列化工具类、编解码类

我看了很多博客，写的大体都差不多这个样子，这里不需要要改动

ObjSerializationUtil

package protostuff;

import com.dyuproject.protostuff.LinkedBuffer;
import com.dyuproject.protostuff.ProtostuffIOUtil;
import com.dyuproject.protostuff.Schema;
import com.dyuproject.protostuff.runtime.RuntimeSchema;
import org.objenesis.Objenesis;
import org.objenesis.ObjenesisStd;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author CBeann
 * @create 2019-09-24 20:01
 */
public class ObjSerializationUtil {

    private static Map<Class<?>, Schema<?>> cachedSchema = new ConcurrentHashMap<>();

    private static Objenesis objenesis = new ObjenesisStd();

    private ObjSerializationUtil() {

    }

    /**
     * 序列化(对象 -> 字节数组)
     *
     * @param obj 对象
     * @return 字节数组
     */
    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();
        }
    }

    /**
     * 反序列化(字节数组 -> 对象)
     *
     * @param data
     * @param cls
     * @param <T>
     */
    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);
        }
    }

    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;
    }

}

ObjDecoder

package protostuff;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;

import java.util.List;

/**
 * @author CBeann
 * @create 2019-09-24 19:53
 */
public class ObjDecoder extends ByteToMessageDecoder {



    private Class<?> genericClass;

    public ObjDecoder(Class<?> genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
        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(ObjSerializationUtil.deserialize(data, genericClass));
    }

}

ObjEncoder

package protostuff;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;

/**
 * @author CBeann
 * @create 2019-09-24 20:03
 */
public class ObjEncoder  extends MessageToByteEncoder {

    private Class<?> genericClass;

    public ObjEncoder(Class<?> genericClass) {
        this.genericClass = genericClass;
    }

    @Override
    protected void encode(ChannelHandlerContext ctx, Object in, ByteBuf out)  {
        if (genericClass.isInstance(in)) {
            byte[] data = ObjSerializationUtil.serialize(in);
            out.writeInt(data.length);
            out.writeBytes(data);
        }
    }

}

修改TimeServer和TimeClient

修改的一样

//对象传输处理
socketChannel.pipeline().addLast(new ObjDecoder(MsgInfo.class));
socketChannel.pipeline().addLast(new ObjEncoder(MsgInfo.class));

修改TimeClient发送POJO

 //发送数据
            MsgInfo msgInfo =
                    new MsgInfo(1, "客户端：--->" + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
            f.channel().writeAndFlush(msgInfo);
            Thread.sleep(4000);
            msgInfo = new MsgInfo(1, "客户端：--->" + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
            f.channel().writeAndFlush(msgInfo);

修改TimeServerHandler

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

        //服务器读客户端发送来的数据
        MsgInfo msgInfo = (MsgInfo) msg;
        System.out.println("The TimeServer receive :" + msgInfo);

        //服务器向客户端回应请求
        msgInfo = new MsgInfo(2, "服务端：--->" + (new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())));
        ctx.writeAndFlush(msgInfo);


    }

修改TimeClientHandler

    //客户端读取服务器发送的数据
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        try {
            //服务器读客户端发送来的数据
            MsgInfo msgInfo = (MsgInfo) msg;
            System.out.println("客户端收到 :   " + msg);
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            //标配
            ReferenceCountUtil.release(msg);
        }
    }