Netty 编解码器学习记：从粘包拆包到自定义协议

这是一个小白的学习记录
边学边练，把踩过的坑都记下来

为啥要学编解码器？

学了线程模型，感觉 Netty 跑起来挺快的。但要想数据传输稳定，必须学会编解码器。

我一开始还纳闷，编解码器不就是序列化和反序列化吗？有啥好学的？后来才发现，Netty 的编解码器还真不简单，特别是粘包/拆包问题，不学还真搞不定。

一、编解码器是啥？

我理解的编解码器

编解码器就是把数据在网络传输格式和应用程序格式之间转换的工具。

  • 编码器（Encoder）：把应用程序的数据转换成网络能传输的格式
  • 解码器（Decoder）：把网络传输的格式转换成应用程序能理解的格式

二、常用编解码器

1. 字符串编解码器

这是最常用的，把字节数据转换成字符串。

ch.pipeline().addLast(new StringDecoder(StandardCharsets.UTF_8));
ch.pipeline().addLast(new StringEncoder(StandardCharsets.UTF_8));

2. 基于长度的帧解码器

这个是解决粘包/拆包问题的神器。

LengthFieldBasedFrameDecoder
// 长度字段占 4 字节，位于消息开头
LengthFieldBasedFrameDecoder decoder = new LengthFieldBasedFrameDecoder(
    1024, // 最大帧长度
    0,    // 长度字段偏移量
    4,    // 长度字段长度
    0,    // 长度调整值
    4     // 跳过长度字段
);
FixedLengthFrameDecoder

用于固定长度的消息。

// 每个消息固定 100 字节
FixedLengthFrameDecoder decoder = new FixedLengthFrameDecoder(100);
LineBasedFrameDecoder

用于基于换行符的消息。

// 最大行长度 1024 字节
LineBasedFrameDecoder decoder = new LineBasedFrameDecoder(1024);

3. 对象编解码器

用于传输 Java 对象。

ch.pipeline().addLast(new ObjectDecoder(ClassResolvers.cacheDisabled(null)));
ch.pipeline().addLast(new ObjectEncoder());

4. Protobuf 编解码器

用于传输 Protobuf 对象，性能更好。

ch.pipeline().addLast(new ProtobufDecoder(MyMessage.getDefaultInstance()));
ch.pipeline().addLast(new ProtobufEncoder());

三、粘包/拆包问题

什么是粘包/拆包？

TCP 是面向流的协议，数据会被分割或合并，导致：

  • 粘包：多个小数据包被合并成一个大数据包
  • 拆包：一个大数据包被分割成多个小数据包

我踩过的坑

刚开始写 Echo 服务器时，客户端发"Hello"，服务端收到"HelloHelloHello"，或者收到"Hel"、“lo”。我当时就懵了，这是啥情况？

后来查资料才明白，这就是粘包/拆包问题。

解决方案

1. 基于长度字段

在消息开头加一个长度字段，告诉接收方消息有多长。

// 编码器：写入长度和数据
ch.pipeline().addLast(new MessageToByteEncoder<String>() {
    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        byte[] data = msg.getBytes();
        out.writeInt(data.length); // 写入长度
        out.writeBytes(data); // 写入数据
    }
});

// 解码器：根据长度字段解析
ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
    1024, 0, 4, 0, 4
));
2. 基于分隔符

用特定的分隔符标记消息结束，比如换行符。

// 编码器：加换行符
ch.pipeline().addLast(new MessageToByteEncoder<String>() {
    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        out.writeBytes((msg + "\n").getBytes());
    }
});

// 解码器：按换行符分割
ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
3. 固定长度

每个消息长度固定，不够的填充。

// 编码器：填充空格
ch.pipeline().addLast(new MessageToByteEncoder<String>() {
    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        String paddedMsg = String.format("%-100s", msg);
        out.writeBytes(paddedMsg.getBytes());
    }
});

// 解码器：固定长度
ch.pipeline().addLast(new FixedLengthFrameDecoder(100));

教训：粘包/拆包问题一定要处理，否则数据传输会乱套！

四、自定义编解码器

实现自定义 Decoder

public class CustomDecoder extends ByteToMessageDecoder {
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 确保有足够的字节可读
        if (in.readableBytes() < 4) {
            return; // 不够，等更多数据
        }
        
        // 读取长度
        int length = in.readInt();
        if (in.readableBytes() < length) {
            in.resetReaderIndex(); // 重置读索引
            return;
        }
        
        // 读取数据
        byte[] data = new byte[length];
        in.readBytes(data);
        String message = new String(data);
        out.add(message);
    }
}

实现自定义 Encoder

public class CustomEncoder extends MessageToByteEncoder<String> {
    @Override
    protected void encode(ChannelHandlerContext ctx, String msg, ByteBuf out) throws Exception {
        byte[] data = msg.getBytes();
        out.writeInt(data.length); // 写入长度
        out.writeBytes(data); // 写入数据
    }
}

五、实战：实现自定义协议

协议设计

我设计了一个简单的协议：

  • 魔数（4字节）：0xCAFEBABE
  • 版本（1字节）：1
  • 命令（1字节）：1-登录，2-消息，3-退出
  • 长度（4字节）：数据部分的长度
  • 数据（可变长度）：具体业务数据

实现编解码器

解码器
public class CustomProtocolDecoder extends ByteToMessageDecoder {
    private static final int MAGIC_NUMBER = 0xCAFEBABE;
    private static final int HEADER_LENGTH = 4 + 1 + 1 + 4; // 魔数 + 版本 + 命令 + 长度
    
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        // 检查是否有足够的字节
        if (in.readableBytes() < HEADER_LENGTH) {
            return;
        }
        
        // 标记当前位置
        in.markReaderIndex();
        
        // 读取魔数
        int magic = in.readInt();
        if (magic != MAGIC_NUMBER) {
            ctx.close(); // 魔数不匹配，关闭连接
            return;
        }
        
        // 读取版本和命令
        byte version = in.readByte();
        byte command = in.readByte();
        
        // 读取数据长度
        int length = in.readInt();
        
        // 检查数据长度
        if (in.readableBytes() < length) {
            in.resetReaderIndex(); // 重置，等更多数据
            return;
        }
        
        // 读取数据
        byte[] data = new byte[length];
        in.readBytes(data);
        
        // 构造消息
        CustomMessage message = new CustomMessage(version, command, data);
        out.add(message);
    }
}
编码器
public class CustomProtocolEncoder extends MessageToByteEncoder<CustomMessage> {
    private static final int MAGIC_NUMBER = 0xCAFEBABE;
    
    @Override
    protected void encode(ChannelHandlerContext ctx, CustomMessage msg, ByteBuf out) throws Exception {
        // 写入魔数
        out.writeInt(MAGIC_NUMBER);
        // 写入版本和命令
        out.writeByte(msg.getVersion());
        out.writeByte(msg.getCommand());
        // 写入数据长度
        out.writeInt(msg.getData().length);
        // 写入数据
        out.writeBytes(msg.getData());
    }
}

使用编解码器

public class CustomProtocolServer {
    public static void main(String[] args) throws Exception {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 public void initChannel(SocketChannel ch) throws Exception {
                     ch.pipeline().addLast(new CustomProtocolDecoder());
                     ch.pipeline().addLast(new CustomProtocolEncoder());
                     ch.pipeline().addLast(new CustomProtocolHandler());
                 }
             });
            
            ChannelFuture f = b.bind(8080).sync();
            System.out.println("Server started on port 8080");
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

六、我踩过的坑

  1. 粘包/拆包：刚开始没处理，导致数据错乱
  2. 长度字段配置错误：LengthFieldBasedFrameDecoder 参数配置错了，导致解码失败
  3. 内存溢出：没设置最大帧长度，导致大消息撑爆内存
  4. 解码器顺序：编解码器顺序放错了，导致数据处理失败

七、编解码器最佳实践

  1. 选择合适的编解码器：根据协议类型选择
  2. 正确配置参数：特别是 LengthFieldBasedFrameDecoder 的参数
  3. 设置最大帧长度：防止内存溢出
  4. 处理边界情况：比如数据不完整、长度不合法
  5. 测试编解码器：多测试正常、边界、异常情况
  6. 避免耗时操作：编解码器里别做耗时的事

验证步骤

1. 测试粘包/拆包

// 客户端连续发送多条消息
for (int i = 0; i < 10; i++) {
    channel.writeAndFlush("Message " + i);
}

预期结果：服务端收到 10 条完整的消息，没有粘在一起。

2. 测试自定义协议

// 发送登录命令
CustomMessage loginMsg = new CustomMessage((byte) 1, (byte) 1, "user:admin,pwd:123".getBytes());
channel.writeAndFlush(loginMsg);

// 发送消息命令
CustomMessage msgMsg = new CustomMessage((byte) 1, (byte) 2, "Hello Netty".getBytes());
channel.writeAndFlush(msgMsg);

预期结果：服务端正确解析出登录和消息命令。

总结

其实 Netty 的编解码器也没那么难，就是把数据在不同格式之间转换，还要处理粘包/拆包问题。我也是踩了几个坑才明白这些道理的。

现在我对编解码器有了点感觉，知道什么时候用什么编解码器，怎么处理粘包/拆包问题。但要真正掌握，还得继续练习。

肯定有理解不对的地方，欢迎大佬指正。

如果你也是新手，希望这篇笔记能帮到你。

# 学习 # java
Logo
https://edu.csdn.net/learn/39067/627173?utm_source=2019755004

汇聚全球AI编程工具，助力开发者即刻编程。

更多推荐

cover

四天蒸发近2000亿，宇树科技股价近乎“腰斩”；苹果举办库克告别派对：9月1日正式卸任CEO；Anthropic两款新Claude模型曝光 | 极客头条

avatar AI编程社区
cover

Claude Code反超GitHub Copilot登顶第一、90%程序员已用上Agent，最新AI编码调查报告来了

avatar AI编程社区
cover

如何用好 Claude Code 写后端代码

avatar AI编程社区