MessagePack 对象序列化、编解码及netty实例开发

冯文彬

2023-12-01

MessagePack 是一个高效的二进制序列化框架，它像json一样支持不同语言间的数据交换，但是它的性能更快，序列化之后的码流也更小

介绍

MessagePack 特点如下：

编解码高效，性能高
序列化之后的码流小
支持跨语言

多语言支持

官方支持的语言如下：Java、Python、Ruby、Haskell、C#、OCaml、Lua、Go、C、C++等。

netty 对象编解码实例

MessagePack 解码器

public class MsgpackEncoder extends MessageToByteEncoder<Object>
{
    @Override
    protected void encode(ChannelHandlerContext ctx , Object o , ByteBuf byteBuf) throws Exception
    {
        MessagePack msgpack=new MessagePack();
        byte[] raw=msgpack.write(o);
        byteBuf.writeBytes(raw);
    }
}

MessagePack 译码器

public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf>
{
    @Override
    protected void decode(ChannelHandlerContext ctx , ByteBuf byteBuf , List<Object> list) throws Exception
    {
        final int length=byteBuf.readableBytes();
        final byte[] array=new byte[length];
        byteBuf.readBytes(array);
        MessagePack msgpack=new MessagePack();
        list.add(msgpack.read(array));
    }
}

客户端添加编码器

在客户端添加编码器，将写入的Object 序列化编码为ByteBuf。

b.group(group)
    .channel(NioSocketChannel.class)
    .handler(new ChannelInitializer<SocketChannel>()
    {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception
        {
            // ch.pipeline().addLast(new LengthFieldPrepender(2));
            ch.pipeline().addLast(new MsgpackEncoder());
            ch.pipeline().addLast(new ClientHandler());
        }
    });

服务端添加解码器

b.group(bossGroup,workerGroup)
    .channel(NioServerSocketChannel.class)
    .childHandler(new ChannelInitializer<SocketChannel>()
    {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception
        {
            // ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
            ch.pipeline().addLast(new MsgpackDecoder());
            ch.pipeline().addLast(new EchoServerHandler());
        }
    })

客户端Handler

@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception
    {
        for (int i = 0; i < 5; i++)
        {
            User u=new User();
            u.setAge(11);
            u.setName("miao");
            ctx.writeAndFlush(u);
            System.out.println("client send object : "+u);
        }
    }

注意
别忘了在 User (你要序列化的类)上添加 @Message 注解

服务端Handler

@Override
    public void channelRead(ChannelHandlerContext ctx,Object msg){
        List<User> u= (List<User>) msg;
        System.out.println("server receive object:"+u);
    }

结果展示

## Client ：
client send object : User(age=11, name=miao)
client send object : User(age=11, name=miao)
client send object : User(age=11, name=miao)
client send object : User(age=11, name=miao)
client send object : User(age=11, name=miao)

## Server :
server receive object:[11,"miao"]
server receive object:[11,"miao"]
server receive object:[11,"miao"]
server receive object:[11,"miao"]
server receive object:[11,"miao"]

粘包拆包问题解决

关于粘包拆包可以参考这篇博客》》》》》

注意之前代码中注释了两条代码：

## 客户端中
ch.pipeline().addLast(new LengthFieldPrepender(2));
## 添加一个 LengthFieldPrepender 编码器，它会在 
## ByteBuf 之前增加两个字节的字段，用于记录消息长度。

## 服务端中
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,2,0,2));
## 添加一个 LengthFieldBasedFrameDecoder 解码器，它会在解码时
## 按照消息头的长度来进行解码。