TCP 粘包与拆包

TCP是一个“流”协议，TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把多个小包封装成一个大的数据包进行发送，这就是TCP的粘包和拆包问题。

TCP 粘包与拆包问题说明

假设客户端分别发送了俩个数据包D1和D2到服务端。但服务端每次读到的字节数是不确定的，所以可能存在下面四种情况：

1. 服务端分两次读取到了D1和D2数据包，也就是理想的正常情况，没有发生粘包和拆包。
2. 服务端一次接收收到了两个数据包，此时，D1和D2粘到了一起，这种情况被称为TCP粘包。
3. 服务端分两次读取到了两个数据包，第一次读取到了D1完整的数据包和D2部分的数据，第二次读取到了D2剩余的数据，这种情况被称为TCP拆包。
4. 服务端分两次读取到了两个数据包，第一次读取到了D2部分数据，第二次读取到了D1剩余的数据和D2完整的数据包，这种情况和第三种情况一样是TCP拆包。
5. 第五种情况要假设TCP接收滑窗非常小，而数据包D1和D2比较大，这样就有可能发生服务端多次接收才能完成接收，期间会发生多次拆包。

TCP 粘包与拆包发生的原因

问题产生的原因有三点：

• 程序写入的字节大小大于套接口发送缓冲去大小。
• 进行MSS大小的TCP分段。
• 以太网帧的payload大于MTU进行IP分片。
  

Netty中解决TCP粘包与拆包问题

TCP粘包与拆包异常案例

我们通过一个简单的Netty服务端和客户端例子来复现一下粘包和拆包的异常情况。这个例子中客户端会向服务端发送一个指定的字符串来查询当前时间，服务端接收客户端发送的字符串，如果是合法的查询字符串则返回当前的系统实现，否则返回错误提示。

服务端TimeServer.java

package problem;

import java.util.logging.Logger;

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 problem.handler.TimeServerHandler;

public class TimeServer {

	private static final Logger logger = Logger.getLogger(TimeServer.class.getName());

	public void run(int port) {
		EventLoopGroup bossGroup = new NioEventLoopGroup();
		EventLoopGroup workGroup = new NioEventLoopGroup();
		try {
			ServerBootstrap b = new ServerBootstrap();
			b.group(bossGroup, workGroup).channel(NioServerSocketChannel.class).option(ChannelOption.SO_BACKLOG, 1024)
					.childOption(ChannelOption.SO_KEEPALIVE, true)
					.childHandler(new ChannelInitializer<SocketChannel>() {
						@Override
						protected void initChannel(SocketChannel ch) throws Exception {
							// TODO Auto-generated method stub
							ch.pipeline().addLast(new TimeServerHandler());
						}
					});
			ChannelFuture f = b.bind(port).sync();
			f.channel().closeFuture().sync();
		} catch (InterruptedException e) {
			logger.warning(e.getLocalizedMessage());
		} finally {
			bossGroup.shutdownGracefully();
			workGroup.shutdownGracefully();
		}
	}

	public static void main(String[] args) {
		new TimeServer().run(8080);
	}
}

服务端处理类TimeServerHandler.java

package problem.handler;

import java.nio.charset.StandardCharsets;
import java.util.Date;
import java.util.logging.Logger;

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

public class TimeServerHandler extends ChannelInboundHandlerAdapter {

	private static final Logger logger = Logger.getLogger(TimeServerHandler.class.getName());

	private int counter;

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		// TODO Auto-generated method stub
		ByteBuf buf = (ByteBuf) msg;
		byte[] req = new byte[buf.readableBytes()];
		buf.readBytes(req);
		String body = new String(req, StandardCharsets.UTF_8).substring(0,
				req.length - System.getProperty("line.separator").length());
		System.out.println("The time server receive order : " + body + "; This counter is : " + (++counter));
		String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
				? new Date(System.currentTimeMillis()).toString() + System.getProperty("line.separator")
				: "BAD ORDERE";
		ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
		ctx.writeAndFlush(resp);
	}

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

}

客户端TimeClient.java

package problem;

import java.util.logging.Logger;

import io.netty.bootstrap.Bootstrap;
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.NioSocketChannel;
import problem.handler.TimeClientHandler;

public class TimeClient {

	private static final Logger logger = Logger.getLogger(TimeClient.class.getName());

	public void run(String host, int port) {
		EventLoopGroup group = new NioEventLoopGroup();
		try {
			Bootstrap b = new Bootstrap();
			b.group(group).channel(NioSocketChannel.class).option(ChannelOption.SO_KEEPALIVE, true)
					.handler(new ChannelInitializer<SocketChannel>() {
						@Override
						protected void initChannel(SocketChannel ch) throws Exception {
							// TODO Auto-generated method stub
							ch.pipeline().addLast(new TimeClientHandler());
						}
					});
			ChannelFuture future = b.connect(host, port).sync();
			future.channel().closeFuture().await(5000L);
		} catch (InterruptedException e) {
			logger.warning(e.getLocalizedMessage());
		} finally {
			group.shutdownGracefully();
		}
	}

	public static void main(String[] args) {
		new TimeClient().run("127.0.0.1", 8080);
	}
}

客户端处理类TimeClientHandler.java

package problem.handler;

import java.nio.charset.StandardCharsets;
import java.util.logging.Logger;

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

public class TimeClientHandler extends ChannelInboundHandlerAdapter {

	private static final Logger logger = Logger.getLogger(TimeClientHandler.class.getName());

	private byte[] req;

	private int counter;

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

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		System.out.println("client connect server successful!");
		ByteBuf message = null;
		for (int i = 0; i < 100; i++) {
			message = Unpooled.buffer(req.length);
			message.writeBytes(req);
			ctx.writeAndFlush(message);
		}
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		System.out.println("读取");
		ByteBuf buf = (ByteBuf) msg;
		byte[] receiveByte = new byte[buf.readableBytes()];
		buf.readBytes(receiveByte);
		String body = new String(receiveByte, StandardCharsets.UTF_8);
		System.out.println("Now is " + body + "; counter:" + (++counter));
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		System.err.println(cause.getLocalizedMessage());
		ctx.close();
	}
}

运行效果

服务端

可以看到服务端只接收了两次，这里发生了粘包

客户端

同样客户端应该收到两条回复才对，但是这里只有一条，也发生了粘包。由于我们程序中没有考虑粘包和拆包所以发生了上面的情况。

使用Netty中的编码器解决TCP粘包问题

为了解决TCP粘包和拆包导致的半包读写问题，Netty提供了一些自带的编码器用来处理半包。下面的代码是对上面发生异常的代码做的修改。

TimeServer.java下在childHandler方法下做如下修改

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new TimeServerHandler());

也就是在原有的基础上增加了LineBasedFrameDecoder这个解码器和StringDecoder解码器。

TimeServerHandler.java下将channelRead这个方法做如下修改

	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		// TODO Auto-generated method stub
//		ByteBuf buf = (ByteBuf) msg;
//		byte[] req = new byte[buf.readableBytes()];
//		buf.readBytes(req);
//		String body = new String(req, StandardCharsets.UTF_8).substring(0,
//				req.length - System.getProperty("line.separator").length());
		String body = (String) msg;
		System.out.println("The time server receive order : " + body + "; This counter is : " + (++counter));
		String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
				? new Date(System.currentTimeMillis()).toString() + System.getProperty("line.separator")
				: "BAD ORDERE";
		ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
		ctx.writeAndFlush(resp);
	}

可以看到，将上面转化接收数据的过程注释掉，然后把msg强转为String即可（相当简洁）。

TimeClient.java在handler方法中做如下修改

ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
ch.pipeline().addLast(new StringDecoder());
ch.pipeline().addLast(new TimeClientHandler());

修改方式和服务端一样。

TimeClientHandler在channelRead方法中做如下修改

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
//		System.out.println("读取");
//		ByteBuf buf = (ByteBuf) msg;
//		byte[] receiveByte = new byte[buf.readableBytes()];
//		buf.readBytes(receiveByte);
//		String body = new String(receiveByte, StandardCharsets.UTF_8);
		String body = (String) msg;
		System.out.println("Now is " + body + "; counter:" + (++counter));
	}

同样和服务端一样。

运行效果

服务端

客户端

此时就会发现粘包的问题已经解决了，使用Netty来解决对使用者来说是很方便的，只需要将支持半包的解码的handler添加到ChannelPipeline中即可，不需要额外的代码。

原理分析

LineBasedFrameDecoder的工作原理是它依次遍历ByteBuf中的可读字节，判断是否有"\n"或者"\r\n"，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，同时支持配置单行最大长度，如果连续读取到最大长度后没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。
StringDecoder的作用就是将接收到的对象转为字符串，然后再继续调用后面的handler。LineBasedFrameDecoder+StringDecoder组合就是按行切换的文本解码器。初次之外，Netty还有其他多种解码器，用来满足不同的需求。

在进行Java NIO学习时，发现，如果客户端连续不断的向服务端发送数据包时，服务端接收的数据会出现两个数据包粘在一起的情况，这就是TCP协议中经常会遇到的粘包以及拆包的问题。

我们都知道TCP属于传输层的协议，传输层除了有TCP协议外还有UDP协议。那么UDP是否会发生粘包或拆包的现象呢？答案是不会。UDP是基于报文发送的，从UDP的帧结构可以看出，在UDP首部采用了16bit来指示UDP数据报文的长度，因此在应用层能很好的将不同的数据报文区分开，从而避免粘包和拆包的问题。而TCP是基于字节流的，虽然应用层和TCP传输层之间的数据交互是大小不等的数据块，但是TCP把这些数据块仅仅看成一连串无结构的字节流，没有边界；另外从TCP的帧结构也可以看出，在TCP的首部没有表示数据长度的字段，基于上面两点，在使用TCP传输数据时，才有粘包或者拆包现象发生的可能。

粘包、拆包表现形式

现在假设客户端向服务端连续发送了两个数据包，用packet1和packet2来表示，那么服务端收到的数据可以分为三种，现列举如下：

第一种情况，接收端正常收到两个数据包，即没有发生拆包和粘包的现象，此种情况不在本文的讨论范围内。

第二种情况，接收端只收到一个数据包，由于TCP是不会出现丢包的，所以这一个数据包中包含了发送端发送的两个数据包的信息，这种现象即为粘包。这种情况由于接收端不知道这两个数据包的界限，所以对于接收端来说很难处理。

第三种情况，这种情况有两种表现形式，如下图。接收端收到了两个数据包，但是这两个数据包要么是不完整的，要么就是多出来一块，这种情况即发生了拆包和粘包。这两种情况如果不加特殊处理，对于接收端同样是不好处理的。

粘包、拆包发生原因

发生TCP粘包或拆包有很多原因，现列出常见的几点，可能不全面，欢迎补充，

1、要发送的数据大于TCP发送缓冲区剩余空间大小，将会发生拆包。

2、待发送数据大于MSS（最大报文长度），TCP在传输前将进行拆包。

3、要发送的数据小于TCP发送缓冲区的大小，TCP将多次写入缓冲区的数据一次发送出去，将会发生粘包。

4、接收数据端的应用层没有及时读取接收缓冲区中的数据，将发生粘包。

等等。

粘包、拆包解决办法

通过以上分析，我们清楚了粘包或拆包发生的原因，那么如何解决这个问题呢？解决问题的关键在于如何给每个数据包添加边界信息，常用的方法有如下几个：

1、发送端给每个数据包添加包首部，首部中应该至少包含数据包的长度，这样接收端在接收到数据后，通过读取包首部的长度字段，便知道每一个数据包的实际长度了。

2、发送端将每个数据包封装为固定长度（不够的可以通过补0填充），这样接收端每次从接收缓冲区中读取固定长度的数据就自然而然的把每个数据包拆分开来。

3、可以在数据包之间设置边界，如添加特殊符号，这样，接收端通过这个边界就可以将不同的数据包拆分开。

等等。

样例程序

我将在程序中使用两种方法来解决粘包和拆包问题，固定数据包长度和添加长度首部，这两种方法各有优劣。固定数据包长度传输效率一般，尤其是在要发送的数据长度长短差别很大的时候效率会比较低，但是编程实现比较简单；添加长度首部虽然可以获得较高的传输效率，冗余信息少且固定，但是编程实现较为复杂。下面给出的样例程序是基于之前的文章《Java中BIO，NIO和AIO使用样例》中提到的NIO实例的，如果对NIO的使用还不是很熟悉，可以先了解一下Java中NIO编程。

固定数据包长度

这种处理方式的思路很简单，发送端在发送实际数据前先把数据封装为固定长度，然后在发送出去，接收端接收到数据后按照这个固定长度进行拆分即可。发送端程序如下：

1. // 发送端

2. String msg = "hello world " + number++;

3. socketChannel.write(ByteBuffer.wrap(new FixLengthWrapper(msg).getBytes()));

4.  

5. // 封装固定长度的工具类

6. public class FixLengthWrapper {

7.  

8. public static final int MAX_LENGTH = 32;

9. private byte[] data;

10.  

11. public FixLengthWrapper(String msg) {

12. ByteBuffer byteBuffer = ByteBuffer.allocate(MAX_LENGTH);

13. byteBuffer.put(msg.getBytes());

14. byte[] fillData = new byte[MAX_LENGTH - msg.length()];

15. byteBuffer.put(fillData);

16. data = byteBuffer.array();

17. }

18.  

19. public FixLengthWrapper(byte[] msg) {

20. ByteBuffer byteBuffer = ByteBuffer.allocate(MAX_LENGTH);

21. byteBuffer.put(msg);

22. byte[] fillData = new byte[MAX_LENGTH - msg.length];

23. byteBuffer.put(fillData);

24. data = byteBuffer.array();

25. }

26.  

27. public byte[] getBytes() {

28. return data;

29. }

30.  

31. public String toString() {

32. StringBuilder sb = new StringBuilder();

33. for (byte b : getBytes()) {

34. sb.append(String.format("0x%02X ", b));

35. }

36. return sb.toString();

37. }

38. }

可以看到客户端在发送数据前首先把数据封装为长度为32bytes的数据包，这个长度是根据目前实际数据包长度来规定的，这个长度必须要大于所有可能出现的数据包的长度，这样才不会出现把数据“截断”的情况。接收端程序如下：

1. private static void processByFixLength(SocketChannel socketChannel) throws IOException {

2. while (socketChannel.read(byteBuffer) > 0) {

3.  

4. byteBuffer.flip();

5. while (byteBuffer.remaining() >= FixLengthWrapper.MAX_LENGTH) {

6. byte[] data = new byte[FixLengthWrapper.MAX_LENGTH];

7. byteBuffer.get(data, 0, FixLengthWrapper.MAX_LENGTH);

8. System.out.println(new String(data) + " <---> " + number++);

9. }

10. byteBuffer.compact();

11. }

12. }

可以看出接收端的处理很简单，只需要每次读取固定的长度即可区分出来不同的数据包。

添加长度首部

这种方式的处理较上面提到的方式稍微复杂一点。在发送端需要给待发送的数据添加固定的首部，然后再发送出去，然后在接收端需要根据这个首部的长度信息进行数据包的组合或拆分，发送端程序如下：

1. // 发送端

2. String msg = "hello world " + number++;

3. // add the head represent the data length

4. socketChannel.write(ByteBuffer.wrap(new PacketWrapper(msg).getBytes()));

5.  

6. // 添加长度首部的工具类

7. public class PacketWrapper {

8.  

9. private int length;

10. private byte[] payload;

11.  

12. public PacketWrapper(String payload) {

13. this.payload = payload.getBytes();

14. this.length = this.payload.length;

15. }

16.  

17. public PacketWrapper(byte[] payload) {

18. this.payload = payload;

19. this.length = this.payload.length;

20. }

21.  

22. public byte[] getBytes() {

23. ByteBuffer byteBuffer = ByteBuffer.allocate(this.length + 4);

24. byteBuffer.putInt(this.length);

25. byteBuffer.put(payload);

26. return byteBuffer.array();

27. }

28.  

29. public String toString() {

30. StringBuilder sb = new StringBuilder();

31. for (byte b : getBytes()) {

32. sb.append(String.format("0x%02X ", b));

33. }

34. return sb.toString();

35. }

36. }

从程序可以看到，发送端在发送数据前首先给待发送数据添加了代表长度的首部，首部长为4bytes（即int型长度），这样接收端在收到这个数据之后，首先需要读取首部，拿到实际数据长度，然后再继续读取实际长度的数据，即实现了组包和拆包的操作。程序如下：

1. private static void processByHead(SocketChannel socketChannel) throws IOException {

2.  

3. while (socketChannel.read(byteBuffer) > 0) {

4. // 保存bytebuffer状态

5. int position = byteBuffer.position();

6. int limit = byteBuffer.limit();

7. byteBuffer.flip();

8. // 判断数据长度是否够首部长度

9. if (byteBuffer.remaining() < 4) {

10. byteBuffer.position(position);

11. byteBuffer.limit(limit);

12. continue;

13. }

14. // 判断bytebuffer中剩余数据是否足够一个包

15. int length = byteBuffer.getInt();

16. if (byteBuffer.remaining() < length) {

17. byteBuffer.position(position);

18. byteBuffer.limit(limit);

19. continue;

20. }

21. // 拿到实际数据包

22. byte[] data = new byte[length];

23.  

24. byteBuffer.get(data, 0, length);

25. System.out.println(new String(data) + " <---> " + number++);

26. byteBuffer.compact();

27. }

28. }

关键信息已经在程序中做了注释，可以很明显的感觉到这种方法的处理难度相对于固定长度要大一些，不过这种方式可以获取更大的传输效率。

这里需要提醒各位同学一个问题，由于我在测试的时候采用的是一台机器连续发送数据来模拟高并发的场景，所以在测试的时候会发现服务器端收到的数据包的个数经常会小于包的序号，好像发生了丢包。但经过仔细分析可以发现，这种情况是因为TCP发送缓存溢出导致的丢包，也就是这个数据包根本没有发出来。也就是说，发送端发送数据过快，导致接收端缓存很快被填满，这个时候接收端会把通知窗口设置为0从而控制发送端的流量，这样新到的数据只能暂存在发送端的发送缓存中，当发送缓存溢出后，就出现了我上面提到的丢包，这个问题可以通过增大发送端缓存来缓解这个问题，

socketChannel.socket().setSendBufferSize(102400);  

当然这个话题不在本文的讨论范围，如果有兴趣的同学可以参阅《TCP/IP详解卷一》中的拥塞窗口一章。

关于源码说明，源码默认是把粘包和拆包处理这一部分注释掉了，分别位于NIOTcpServer和NIOTcpClient文件中，需要测试粘包和拆包处理程序的同学需要把这一段注释给去掉。

最后给出源码下载地址

参考

Netty精粹之TCP粘包拆包问题