TCP粘包/拆包--利用LineBasedFrameDecoder解决TCP粘包问题

节选自 Netty权威指南 第二版

TCP是个“流”协议，所谓流，就是没有界面的一串数据。大家可以想象河里的流水，

它们是连成一片的，其间并没有分界线。TCP底层并不了解上层业务数据的具体含义，它

会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会

被TCP拆分成多个包进行发送，也有可能把多个小小的包封装成一个大的数据包发送，这

就是所谓的TCP粘包和拆包问题。

TCP粘包/拆包问题说明

我们可以通过图解对TCP粘包和拆包问题进行说明，如下图



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

1） 服务端分两次取到了两个独立的数据包，分别是D1和D2，没有粘包和拆包。

2） 服务端一次接收到了两个数据包，D1和D2粘合在一起，被称为TCP粘包。

3） 服务端分两次读取到了两个数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次取到了D2包的剩余内容，这被称为TCP粘包。

4） 服务端分两次读取到了两个数据包，第一次读取到D1包的部分内容D1_1，第二次读取到了D1包的剩余内容和D2包的整包。

如果此时服务端TCP接受滑窗非常小，而数据包D1和D2比较大，很有可能会发生第5种可能，即服务端分多次才能将D1和D2包接收完全，期间发生多次拆包。

TCP粘包/拆包发生的原因

问题产生的原因有三个， 分别如下。
1. 应用程序write写入的字节大小大于套接口发送缓冲区大小；
2. 进行MSS大小的TCP分段；
3. 以太网帧的payload大于MTU进行IP分片。

粘包问题的解决策略

由于底层的TCP无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界的主流协议的方案，可以归纳如下。
1. 消息定长，例如每个报文的大小为固定长度200字节，如果不够，空位补空格；
2. 在包尾增加回车换行符进行分割，例如FTP协议；
3. 将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）的字段，通常涉及思路为消息头的第一个字段使用int32来表示消息的总长度；
4. 更复杂的应用层协议。

未考虑TCP粘包功能导致异常案例

1. TimeServer.java

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;

import java.net.InetSocketAddress;

public class TimeServer {
private final static int PORT = 8080;

public static void main(String[] args) {
start();
}

private static void start() {
final TimeServerHandler serverHandler = new TimeServerHandler();
// 创建EventLoopGroup
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
// 创建EventLoopGroup
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
//指定所使用的NIO传输Channel
.channel(NioServerSocketChannel.class)
//使用指定的端口设置套接字地址
.localAddress(new InetSocketAddress(PORT))
// 添加一个EchoServerHandler到Channle的ChannelPipeline
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//EchoServerHandler被标注为@shareable,所以我们可以总是使用同样的案例
socketChannel.pipeline().addLast(serverHandler);
}
});

try {
ChannelFuture f = b.bind().sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

2.TimeServerHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

@Sharable
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
private int counter;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
byte[] req = new byte[buf.readableBytes()];
buf.readBytes(req);
String body = new String(req, "UTF-8").substring(0, req.length - System.getProperty("line.separator").length());
System.out.println(
"The time server receive order: " + body+"; the counter is "+ ++counter
);

String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString()
: "BAD ORDER";

currentTime = currentTime + System.getProperty("line.separator");
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.writeAndFlush(resp);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

3. TimeClient.java

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;

public class TimeClient {
private static final String HOST = "localhost";
private static final int PORT = 8080;

public static void main(String[] args) {
start();
}

private static void start() {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(HOST, PORT))
.handler(new ChannelInitializer<SocketChannel>() {
@Override

4000
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
try {
ChannelFuture future = bootstrap.connect().sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
group.shutdownGracefully();
}
}
}

4. TimeClientHandler.java

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;

public class TimeClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
private int counter;
private byte[] req;
private int req_times = 100;

public TimeClientHandler() {
req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message = null;
for (int i=0; i<req_times; i++){
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
}

@Override
protected void channelRead0(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf) throws Exception {
byte[] req = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(req);
String body = new String(req, "UTF-8");
System.out.println(
"Now is : "+ body +" ; the counter is :" + ++counter
);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

客户端跟服务端链路建立成功之后，循环发送100条消息，每发送一条就刷新一次，保证每条消息都会被写入Channel中。按照我们的设计，服务端应该接收到100条查询时间指令的请求消息。

分别运行服务端和客户端，运行结果如下。

服务端运行结果如下。

------------------------------------------------
The time server receive order: QUERY TIME ORDER

此处省略55行 QUERY TIME ORDER
QUERY TIME ORD; the counter is 1

The time server receive order: 
此处省略42行 QUERY TIME ORDE
QUERY TIME ORD; the counter is 2服务端运行结果表明它只接收到了两条消息，第一条包含57条“QUERY TIME ORDER”指令，第二条包含了43条“QUERY
TIME ORDER”指令，总数正好是100条。而我们期待的是收到100条消息，每条包含一条“QUERY TIME ORDER”指令，这说明发生了TCP粘包。

客户端运行结果如下。

------------------------------------------------
Now is : BAD ORDER

BAD ORDER
 ; the counter is :1

按照设计初衷， 客户端应该收到100条当前系统时间的消息，但实际上只收到了一条。这不难理解，因为服务器端只收到了2条请求消息，所以实际服务端只发送了2条应答，由于请求消息不满足查询条件，所以返回了2条“BAD ORDER”应答消息。但是实际上客户端只收到了一条包含2条“BAD ORDER”指令的消息，说明服务端返回的应答消息也发生了粘包。
由于上面的程序没有考虑TCP的粘包/拆包，所以当发生TCP粘包时，我们的程序就不能正常工作。
下面将演示如何通过Netty的LineBasedFrameDecoder和StringDecoder来解决TCP粘包问题。

利用LineBasedFrameDecoder解决TCP粘包问题

为了解决TCP粘包/拆包导致的半包读写为，Netty默认提供了多种编码器用于处理半包，只要能熟练掌握这些类库的使用，TCP粘包问题从此会变得非常容易，你甚至不需要关心他们，这也是其他NIO框架和JDK原生的NIO API所无法匹敌的。

改进的TimeServer

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

import java.net.InetSocketAddress;

public class TimeServer {
private final static int PORT = 8080;

public static void main(String[] args) {
start();
}

private static void start() {
final TimeServerHandler serverHandler = new TimeServerHandler();
// 创建EventLoopGroup
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
// 创建EventLoopGroup
ServerBootstrap b = new ServerBootstrap();
b.group(bossGroup, workerGroup)
//指定所使用的NIO传输Channel
.channel(NioServerSocketChannel.class)
.option(ChannelOption.SO_BACKLOG, 1024)
//使用指定的端口设置套接字地址
.localAddress(new InetSocketAddress(PORT))
// 添加一个EchoServerHandler到Channle的ChannelPipeline
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
//EchoServerHandler被标注为@shareable,所以我们可以总是使用同样的案例
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(serverHandler);
}
});

try {
ChannelFuture f = b.bind().sync();
f.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
bossGroup.shutdownGracefully();
workerGroup.shutdownGracefully();
}
}
}

改进的TimeServerHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler.Sharable;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import java.util.Date;

@Sharable
public class TimeServerHandler extends ChannelInboundHandlerAdapter {
private int counter;
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String) msg;
System.out.println(
"The time server receive order: " + body+"; the counter is "+ ++counter
);

String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString()
: "BAD ORDER";

currentTime = currentTime + System.getProperty("line.separator");
ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
ctx.writeAndFlush(resp);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

改进的TimeClient

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LineBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;

import java.net.InetSocketAddress;

public class TimeClient {
private static final String HOST = "localhost";
private static final int PORT = 8080;

public static void main(String[] args) {
start();
}

private static void start() {
EventLoopGroup group = new NioEventLoopGroup();
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(group)
.channel(NioSocketChannel.class)
.remoteAddress(new InetSocketAddress(HOST, PORT))
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel socketChannel) throws Exception {
socketChannel.pipeline().addLast(new LineBasedFrameDecoder(1024));
socketChannel.pipeline().addLast(new StringDecoder());
socketChannel.pipeline().addLast(new TimeClientHandler());
}
});
try {
ChannelFuture future = bootstrap.connect().sync();
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
group.shutdownGracefully();
}
}
}

改进的TimeClientHandler

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.CharsetUtil;

public class TimeClientHandler extends ChannelInboundHandlerAdapter {
private int counter;
private byte[] req;
private int req_times = 100;

public TimeClientHandler() {
req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
}

@Override
public void channelActive(ChannelHandlerContext ctx) throws Exception {
ByteBuf message = null;
for (int i=0; i<req_times; i++){
message = Unpooled.buffer(req.length);
message.writeBytes(req);
ctx.writeAndFlush(message);
}
}

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
String body = (String) msg;
System.out.println(
"Now is " + body + " ; the counter is "+ ++counter
);
}

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
cause.printStackTrace();
ctx.close();
}
}

分别运行TimeServer和TimeClient，执行结果如下。
服务端执行结果如下。

------------------------------------------------

The time server receive order: QUERY TIME ORDER; the counter is 1

// 此处省略2-99行
The time server receive order: QUERY TIME ORDER; the counter is 100

客户端运行结果如下。
------------------------------------------------

Now is Mon Oct 23 17:30:07 CST 2017 ; the counter is 1
// 此处省略2-99行
Now is Mon Oct 23 17:30:07 CST 2017 ; the counter is 100

程序的运行结果安全符合预期，说明通过使用LineBasedFrameDecoder和StringDecoder成功解决了TCP粘包导致的读半包问题。对于使用者来说，只要将支持半包解码的Handler添加到ChannelPipeline中即可，不需要写额外的代码，用户使用起来非常简单。

LineBasedFrameDecoder和StringDecoder的原理分析

LineBasedFrameDecoder的工作原理是它一次遍历ByteBuf中的可读字节，判断看是否有“\n”或者“\r\n”， 如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有出现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。
StringDecoder的功能非常简单，就是将接收到的对象转换成字符串，然后继续调用后面的Handler。LineBasedFrameDecoder+StringDecoder组合就是按行切换的文本解码器，它被设计用来支持TCP的粘包和拆包。
当然，如果发送的消息不是以换行符结束的，该怎么办呢？或者没有回车换行符，靠消息头中的长度字段来分包怎么办？是不是需要自己写半包解码器？
答案是否定的，Netty提供了多种支持TCP粘包/拆包的解码器用来满足用户的不同诉求。