TCP的粘包和拆包及Netty中的解决方案

1.基本介绍

TCP 是面向连接的，面向流的，提供高可靠性服务。收发两端(客户端和服务器端)都要有一一成对的 socket，因此，发送端为了将多个发给接收端的包，更有效的发给对方，使用了优化方法(Nagle 算法)，将多次间隔较小且数据量小的数据，合并成一个大的数据块，然后进行封包。这样做虽然提高了效率，但是接收端就难于分辨出完整的数据包了，因为面向流的通信是无消息保护边界的
由于TCP无消息保护边界,需要在接收端处理消息边界问题，也就是我们所说的粘包、拆包问题,看一张图

假设客户端分别发送了两个数据包 D1 和 D2 给服务端，由于服务端一次读取到字节数是不确定的，故可能存在以下四种情况:

服务端分两次读取到了两个独立的数据包，分别是 D1 和 D2，没有粘包和拆包
服务端一次接受到了两个数据包，D1 和 D2 粘合在一起，称之为 TCP 粘包
服务端分两次读取到了数据包，第一次读取到了完整的 D1 包和 D2 包的部分内容，第二次读取到了 D2 包的剩余内容，这称之为 TCP 拆包
服务端分两次读取到了数据包，第一次读取到了 D1 包的部分内容 D1_1，第二次读取到了 D1 包的剩余部分内容 D1_2 和完整的 D2 包

2.解决方案

由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的解决方案，可以归纳如下。

消息长度固定，累计读取到长度和为定长LEN的报文后，就认为读取到了一个完整的信息
将回车换行符作为消息结束符
将特殊的分隔符作为消息的结束标志，回车换行符就是一种特殊的结束分隔符
通过在消息头中定义长度字段来标识消息的总长度

3.Netty中的解决方案

针对上一小节描述的粘包和拆包的解决方案，对于拆包问题比较简单，用户可以自己定义自己的编码器进行处理，Netty并没有提供相应的组件。对于拆包的问题，由于拆包比较复杂，代码处理比较繁琐，Netty提供了4种解码器来解决，分别如下：

固定长度的拆包器 FixedLengthFrameDecoder，每个应用层数据包的都拆分成都是固定长度的大小
行拆包器 LineBasedFrameDecoder，每个应用层数据包，都以换行符作为分隔符，进行分割拆分
分隔符拆包器 DelimiterBasedFrameDecoder，每个应用层数据包，都通过自定义的分隔符，进行分割拆分
基于数据包长度的拆包器 LengthFieldBasedFrameDecoder，将应用层数据包的长度，作为接收端应用层数据包的拆分依据。按照应用层数据包的大小，拆包。这个拆包器，有一个要求，就是应用层协议中包含数据包的长度

4.自定义编解码器演示

对于粘包与拆包问题，其实前面四种基本上已经能够满足大多数情形了，但是对于一些更加复杂的协议，可能有一些定制化的需求。对于这些场景，其实本质上，我们也不需要手动从头开始写一份粘包与拆包处理器，而是通过继承LengthFieldBasedFrameDecoder和LengthFieldPrepender来实现粘包和拆包的处理。

如果用户确实需要不通过继承的方式实现自己的粘包和拆包处理器，这里可以通过实现MessageToByteEncoder和ByteToMessageDecoder来实现。

我们下面用代码演示自定义一个协议包，将每一次请求的内容的length长度和内容本身存起来，然后解码的时候读取指定length长度的编解码器。

MessageProtocol.java

//协议包
public class MessageProtocol {private int len; //关键private byte[] content;
public int getLen() {return len;}
public void setLen(int len) {this.len = len;}
public byte[] getContent() {return content;}
public void setContent(byte[] content) {this.content = content;}
}

客户端或者服务端发送一个自定义协议包MessageProtocol对象。然后通过编码器接收该对象，并分两次发送。第一次将长度length发过去，第二次将文本内容发过去。

MyMessageEncoder.java

public class MyMessageEncoder extends MessageToByteEncoder<MessageProtocol> {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder encode 方法被调用");out.writeInt(msg.getLen());out.writeBytes(msg.getContent());}
}

解码器接收到编码器发送的两次数据。先读取到数据的length，然后可以读取指定length长度的字节，就可以避免粘包拆包的问题了。

这里解码器并没有直接继承ByteToMessageDecoder，而是使用了类ReplayingDecoder。ReplayingDecoder继承自ByteToMessageDecoder。

关于该类的介绍可以看文章：https://www.jianshu.com/p/4cbf8a07b492 这里就不详细介绍了。

MyMessageDecoder.java

public class MyMessageDecoder extends ReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {System.out.println("MyMessageDecoder decode 被调用");//需要将得到二进制字节码-> MessageProtocol 数据包(对象)int length = in.readInt();
byte[] content = new byte[length];in.readBytes(content);
//封装成 MessageProtocol 对象，放入 out， 传递下一个handler业务处理MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);
out.add(messageProtocol);
}
}

下面把客户端和服务端的代码补齐

MyServer.java

public class MyServer {public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer());

ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();
} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}
}
}

MyServerHandler.java

public class MyServer {public static void main(String[] args) throws Exception {
EventLoopGroup bossGroup = new NioEventLoopGroup(1);EventLoopGroup workerGroup = new NioEventLoopGroup();
try {
ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new MyServerInitializer());

ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();channelFuture.channel().closeFuture().sync();
} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}
}
}

MyServerInitializer.java

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();
pipeline.addLast(new MyMessageDecoder());//解码器pipeline.addLast(new MyMessageEncoder());//编码器pipeline.addLast(new MyServerHandler());}
}

MyClient.java

public class MyClient {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 MyClientInitializer()); //自定义一个初始化类
ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
channelFuture.channel().closeFuture().sync();
}finally {group.shutdownGracefully();}}
}

MyClientInitializer.java

public class MyClientInitializer extends ChannelInitializer<SocketChannel> {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {
ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageEncoder()); //加入编码器pipeline.addLast(new MyMessageDecoder()); //加入解码器pipeline.addLast(new MyClientHandler());}
}

MyClientHandler.java

public class MyClientHandler extends SimpleChannelInboundHandler<MessageProtocol> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//使用客户端发送5条数据 "今天天气冷，吃火锅" 编号for(int i = 0; i< 5; i++) {String mes = "今天天气冷，吃火锅";byte[] content = mes.getBytes(Charset.forName("utf-8"));int length = mes.getBytes(Charset.forName("utf-8")).length;//创建协议包对象MessageProtocol messageProtocol = new MessageProtocol();messageProtocol.setLen(length);messageProtocol.setContent(content);ctx.writeAndFlush(messageProtocol);}}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocol msg) throws Exception {int len = msg.getLen();byte[] content = msg.getContent();System.out.println();System.out.println("客户端接收到消息如下");System.out.println("长度=" + len);System.out.println("内容=" + new String(content, Charset.forName("utf-8")));System.out.println("客户端接收消息数量=" + (++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息=" + cause.getMessage());ctx.close();}
}

启动我们的服务端和客户端，控制台打印信息如下：

服务端：

MyMessageDecoder decode 被调用服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=1
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=2
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=3
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=4
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用服务器接收到信息如下
长度=27
内容=今天天气冷，吃火锅
服务器接收到消息包数量=5
MyMessageEncoder encode 方法被调用

客户端:

MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageEncoder encode 方法被调用
MyMessageDecoder decode 被调用客户端接收到消息如下
长度=36
内容=2b62a2df-1c7a-4ecc-b6b2-6706f919d33d
客户端接收消息数量=1
MyMessageDecoder decode 被调用客户端接收到消息如下
长度=36
内容=f844ae51-49b4-410b-a79f-0e2fa59c0290
客户端接收消息数量=2
MyMessageDecoder decode 被调用客户端接收到消息如下
长度=36
内容=66437198-8674-4c04-be70-def9fbf59199
客户端接收消息数量=3
MyMessageDecoder decode 被调用客户端接收到消息如下
长度=36
内容=89500d0e-6eb3-46b6-87e5-0ae0e4c6357e
客户端接收消息数量=4
MyMessageDecoder decode 被调用客户端接收到消息如下
长度=36
内容=8b861694-a61b-488c-95b6-a030204f4f61
客户端接收消息数量=5