Netty教程 / 第 80 节
第8章：粘包与拆包解决方案
本章导读
粘包和拆包是 TCP 网络编程中的常见问题。本章将深入讲解问题产生的原因、多种解决方案的详细实现，以及在实际项目中的最佳实践。
8.1 TCP 粘包/拆包问题原理
8.1.1 问题现象
正常情况：
发送: [D1] [D2] [D3]
接收: [D1] [D2] [D3]
粘包：
发送: [D1] [D2] [D3]
接收: [D1D2D3]
拆包：
发送: [D1大包]
接收: [D1前半] [D1后半]
粘包+拆包：
发送: [D1] [D2大包] [D3]
接收: [D1D2前半] [D2后半D3]
8.1.2 产生原因
发送端原因：
- Nagle 算法：合并小包，减少网络传输次数
- TCP 缓冲区：数据先写入缓冲区，达到一定量才发送
接收端原因：
- TCP 缓冲区：接收的数据先放入缓冲区
- 应用层读取速度：读取速度慢导致数据堆积
网络原因：
- MSS 限制：最大报文段大小限制，大包会被拆分
- MTU 限制：最大传输单元限制
8.1.3 演示代码
// 服务端（未处理粘包/拆包）
public class ProblematicServer {
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ProblematicHandler());
}
});
bootstrap.bind(8080).sync().channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
class ProblematicHandler extends ChannelInboundHandlerAdapter {
private int count = 0;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
ByteBuf buf = (ByteBuf) msg;
byte[] data = new byte[buf.readableBytes()];
buf.readBytes(data);
String message = new String(data, CharsetUtil.UTF_8);
System.out.println("第" + (++count) + "次接收: " + message);
System.out.println("长度: " + data.length);
buf.release();
}
}
// 客户端（发送多个小包）
public class ProblematicClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 快速发送10个小包
for (int i = 0; i < 10; i++) {
String msg = "Message" + i + "\n";
ctx.writeAndFlush(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8));
}
}
});
}
});
bootstrap.connect("localhost", 8080).sync().channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
可能的输出（粘包）：
第1次接收: Message0
Message1
Message2
长度: 27
第2次接收: Message3
Message4
长度: 18
8.2 固定长度解码器（FixedLengthFrameDecoder）
8.2.1 原理
每个消息固定长度，不足的用空格或特殊字符填充。
消息格式：
+----------+----------+----------+
| Message1 | Message2 | Message3 |
| 10字节 | 10字节 | 10字节 |
+----------+----------+----------+
8.2.2 实现
// 服务端
pipeline.addLast(new FixedLengthFrameDecoder(10)); // 每个消息10字节
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast(new FixedLengthHandler());
class FixedLengthHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println("收到消息: [" + msg.trim() + "]");
}
}
// 客户端
public class FixedLengthClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 发送固定长度消息
for (int i = 0; i < 5; i++) {
String msg = String.format("%-10s", "Msg" + i); // 左对齐，10字符
ctx.writeAndFlush(Unpooled.copiedBuffer(msg, CharsetUtil.UTF_8));
}
}
});
}
});
bootstrap.connect("localhost", 8080).sync().channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
优点：
- 实现简单
- 解码效率高
缺点：
- 浪费空间（短消息需要填充）
- 不适合变长消息
8.3 分隔符解码器（DelimiterBasedFrameDecoder）
8.3.1 行分隔符（LineBasedFrameDecoder）
// 服务端
pipeline.addLast(new LineBasedFrameDecoder(1024)); // 最大长度1024
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
pipeline.addLast(new LineHandler());
class LineHandler extends SimpleChannelInboundHandler<String> {
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println("收到消息: " + msg);
}
}
// 客户端
ctx.writeAndFlush(Unpooled.copiedBuffer("Hello\n", CharsetUtil.UTF_8));
ctx.writeAndFlush(Unpooled.copiedBuffer("World\n", CharsetUtil.UTF_8));
8.3.2 自定义分隔符
// 使用 $$ 作为分隔符
ByteBuf delimiter = Unpooled.copiedBuffer("$$", CharsetUtil.UTF_8);
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
// 客户端
ctx.writeAndFlush(Unpooled.copiedBuffer("Message1$$", CharsetUtil.UTF_8));
ctx.writeAndFlush(Unpooled.copiedBuffer("Message2$$", CharsetUtil.UTF_8));
8.3.3 多个分隔符
ByteBuf[] delimiters = new ByteBuf[] {
Unpooled.copiedBuffer("\n", CharsetUtil.UTF_8),
Unpooled.copiedBuffer("$$", CharsetUtil.UTF_8)
};
pipeline.addLast(new DelimiterBasedFrameDecoder(1024, delimiters));
优点：
- 灵活，支持变长消息
- 易于理解和实现
缺点：
- 消息内容不能包含分隔符
- 需要扫描整个消息查找分隔符
8.4 长度字段解码器（LengthFieldBasedFrameDecoder）
8.4.1 基本用法
/**
* 消息格式：
* +--------+----------+
* | Length | Content |
* | 4 byte | N bytes |
* +--------+----------+
*/
pipeline.addLast(new LengthFieldBasedFrameDecoder(
1024, // maxFrameLength: 最大帧长度
0, // lengthFieldOffset: 长度字段偏移量
4, // lengthFieldLength: 长度字段长度
0, // lengthAdjustment: 长度调整值
4 // initialBytesToStrip: 跳过的字节数
));
// 编码器
pipeline.addLast(new LengthFieldPrepender(4));
8.4.2 参数详解
示例1：长度字段在开头，不包含长度字段本身
消息格式：
+--------+----------+
| Length | Content | Length = Content的长度
| 2 byte | N bytes |
+--------+----------+
解码后：
+----------+
| Content |
| N bytes |
+----------+
配置：
new LengthFieldBasedFrameDecoder(1024, 0, 2, 0, 2);
↑ ↑ ↑ ↑
| | | 跳过2字节（长度字段）
| | 长度调整值0
| 长度字段2字节
长度字段从0开始
示例2：长度字段包含自身
消息格式：
+--------+----------+
| Length | Content | Length = Content长度 + 2
| 2 byte | N bytes |
+--------+----------+
配置：
new LengthFieldBasedFrameDecoder(1024, 0, 2, -2, 2);
↑ ↑ ↑
| | 长度调整值-2（减去长度字段）
| 长度字段2字节
长度字段从0开始
示例3：长度字段在中间
消息格式：
+------+--------+----------+
| Type | Length | Content |
| 1 B | 2 B | N bytes |
+------+--------+----------+
解码后：
+----------+
| Content |
| N bytes |
+----------+
配置：
new LengthFieldBasedFrameDecoder(1024, 1, 2, 0, 3);
↑ ↑ ↑ ↑
| | | 跳过3字节（Type+Length）
| | 长度调整值0
| 长度字段2字节
长度字段从第1字节开始
示例4：长度字段包含整个消息
消息格式：
+------+--------+----------+
| Type | Length | Content | Length = Type + Length + Content
| 1 B | 2 B | N bytes |
+------+--------+----------+
配置：
new LengthFieldBasedFrameDecoder(1024, 1, 2, -3, 0);
↑ ↑ ↑ ↑
| | | 不跳过
| | 长度调整值-3
| 长度字段2字节
长度字段从第1字节开始
8.4.3 完整示例
public class LengthFieldServer {
public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// 解码器
pipeline.addLast(new LengthFieldBasedFrameDecoder(
1024, 0, 4, 0, 4));
// 字符串解码
pipeline.addLast(new StringDecoder(CharsetUtil.UTF_8));
// 业务Handler
pipeline.addLast(new SimpleChannelInboundHandler<String>() {
private int count = 0;
@Override
protected void channelRead0(ChannelHandlerContext ctx, String msg) {
System.out.println("第" + (++count) + "次接收: " + msg);
}
});
}
});
ChannelFuture future = bootstrap.bind(8080).sync();
System.out.println("服务器启动成功");
future.channel().closeFuture().sync();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}
public class LengthFieldClient {
public static void main(String[] args) throws Exception {
EventLoopGroup group = new NioEventLoopGroup();
try {
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// 编码器（自动添加长度字段）
pipeline.addLast(new LengthFieldPrepender(4));
// 字符串编码
pipeline.addLast(new StringEncoder(CharsetUtil.UTF_8));
// 发送Handler
pipeline.addLast(new ChannelInboundHandlerAdapter() {
@Override
public void channelActive(ChannelHandlerContext ctx) {
// 快速发送10个消息
for (int i = 0; i < 10; i++) {
ctx.writeAndFlush("Message" + i);
}
}
});
}
});
ChannelFuture future = bootstrap.connect("localhost", 8080).sync();
future.channel().closeFuture().sync();
} finally {
group.shutdownGracefully();
}
}
}
输出（完美解决粘包/拆包）：
服务器启动成功
第1次接收: Message0
第2次接收: Message1
第3次接收: Message2
第4次接收: Message3
第5次接收: Message4
第6次接收: Message5
第7次接收: Message6
第8次接收: Message7
第9次接收: Message8
第10次接收: Message9
8.5 自定义解码器
8.5.1 自定义协议设计
协议格式：
+--------+------+--------+----------+
| Magic | Type | Length | Content |
| 4 byte | 1 B | 4 B | N bytes |
+--------+------+--------+----------+
Magic: 0x12345678（魔数，用于识别协议）
Type: 消息类型
Length: Content的长度
Content: 消息内容
8.5.2 实现
public class CustomProtocolDecoder extends ByteToMessageDecoder {
private static final int HEADER_SIZE = 9; // 4 + 1 + 4
private static final int MAGIC_NUMBER = 0x12345678;
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 1. 检查是否有完整的头部
if (in.readableBytes() < HEADER_SIZE) {
return;
}
// 2. 标记读取位置
in.markReaderIndex();
// 3. 读取魔数
int magic = in.readInt();
if (magic != MAGIC_NUMBER) {
// 魔数不匹配，关闭连接
ctx.close();
throw new RuntimeException("非法的魔数: 0x" + Integer.toHexString(magic));
}
// 4. 读取类型
byte type = in.readByte();
// 5. 读取长度
int length = in.readInt();
// 6. 检查长度是否合法
if (length < 0 || length > 1024 * 1024) { // 最大1MB
ctx.close();
throw new RuntimeException("非法的长度: " + length);
}
// 7. 检查是否有完整的数据
if (in.readableBytes() < length) {
in.resetReaderIndex();
return;
}
// 8. 读取数据
byte[] data = new byte[length];
in.readBytes(data);
// 9. 创建消息对象
CustomMessage msg = new CustomMessage();
msg.setType(type);
msg.setContent(new String(data, CharsetUtil.UTF_8));
out.add(msg);
}
}
public class CustomProtocolEncoder extends MessageToByteEncoder<CustomMessage> {
private static final int MAGIC_NUMBER = 0x12345678;
@Override
protected void encode(ChannelHandlerContext ctx, CustomMessage msg, ByteBuf out) {
byte[] content = msg.getContent().getBytes(CharsetUtil.UTF_8);
// 写入魔数
out.writeInt(MAGIC_NUMBER);
// 写入类型
out.writeByte(msg.getType());
// 写入长度
out.writeInt(content.length);
// 写入内容
out.writeBytes(content);
}
}
class CustomMessage {
private byte type;
private String content;
// getter/setter...
}
8.6 最佳实践
8.6.1 方案选择
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 固定长度消息 | FixedLengthFrameDecoder | 简单高效 |
| 文本协议 | LineBasedFrameDecoder | 易于调试 |
| 变长消息 | LengthFieldBasedFrameDecoder | 灵活高效 |
| 自定义协议 | 自定义解码器 | 完全控制 |
| HTTP/WebSocket | 内置解码器 | 标准协议 |
8.6.2 性能优化
// 1. 合理设置最大帧长度
new LengthFieldBasedFrameDecoder(
10 * 1024 * 1024, // 10MB，根据实际情况调整
0, 4, 0, 4
);
// 2. 使用池化ByteBuf
bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);
// 3. 批量发送
List<String> messages = ...;
for (String msg : messages) {
ctx.write(msg); // 不立即刷新
}
ctx.flush(); // 批量刷新
8.6.3 安全考虑
public class SecureDecoder extends ByteToMessageDecoder {
private static final int MAX_FRAME_LENGTH = 10 * 1024 * 1024; // 10MB
@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
// 1. 检查长度
if (in.readableBytes() < 4) {
return;
}
in.markReaderIndex();
int length = in.readInt();
// 2. 防止恶意超大包
if (length > MAX_FRAME_LENGTH) {
ctx.close();
throw new TooLongFrameException("帧长度超过限制: " + length);
}
// 3. 防止负数长度
if (length < 0) {
ctx.close();
throw new CorruptedFrameException("非法的长度: " + length);
}
// 继续解码...
}
}
8.7 本章小结
本章我们深入学习了粘包/拆包问题：
✅ 问题原理：TCP 流式协议导致的消息边界问题
✅ 固定长度：FixedLengthFrameDecoder
✅ 分隔符：LineBasedFrameDecoder、DelimiterBasedFrameDecoder
✅ 长度字段：LengthFieldBasedFrameDecoder（推荐）
✅ 自定义协议：自定义解码器
关键要点
- 粘包/拆包是 TCP 的固有特性，应用层必须处理
- LengthFieldBasedFrameDecoder 是最常用和推荐的方案
- 自定义协议要考虑魔数、版本、长度限制
- 要防止恶意超大包攻击
- 编解码器要放在 Pipeline 的最前面
上一章：第7章：编解码器
下一章：第9章：常用协议支持