三、Netty的粘包半包问题解决

一、定义

TCP 传输中，客户端发送数据，实际是把数据写入到了 TCP 的缓存中，粘包和半包也就会在此时产生。客户端给服务端发送了两条消息ABC和DEF，服务端这边的接收会有多少种情况呢？有可能是一次性收到了所有的消息ABCDEF，有可能是收到了三条消息AB、CD、EF。

粘包

上面所说的一次性收到了所有的消息ABCDEF，类似于粘包。如果客户端发送的包的大小比 TCP 的缓存容量小，并且 TCP 缓存可以存放多个包，那么客户端和服务端的一次通信就可能传递了多个包，这时候服务端从 TCP 缓存就可能一下读取了多个包，这种现象就叫粘包。

半包

上面说的后面那种收到了三条消息AB、CD、EF，类似于半包。如果客户端发送的包的大小比 TCP 的缓存容量大，那么这个数据包就会被分成多个包，通过 Socket 多次发送到服务端，服务端第一次从接受缓存里面获取的数据，实际是整个包的一部分，这时候就产生了半包(半包不是说只收到了全包的一半，是说收到了全包的一部分)。

二、产生原因

其实从上面的定义，我们就可以大概知道产生的原因了。

粘包的主要原因：

发送方每次写入数据 < 套接字(Socket)缓冲区大小

接收方读取套接字(Socket)缓冲区数据不够及时

半包的主要原因：

发送方每次写入数据 > 套接字(Socket)缓冲区大小

发送的数据大于协议的 MTU (Maximum Transmission Unit，最大传输单元)，因此必须拆包

其实我们可以换个角度看待问题：

从收发的角度看，便是一个发送可能被多次接收，多个发送可能被一次接收。

从传输的角度看，便是一个发送可能占用多个传输包，多个发送可能共用一个传输包。

根本原因，其实是

TCP 是流式协议，消息无边界。

(PS ： UDP虽然也可以一次传输多个包或者多次传输一个包，但每个消息都是有边界的，因此不会有粘包和半包问题。)

三、现象分析

滑动窗口

TCP 以一个段（segment）为单位，每发送一个段就需要进行一次确认应答（ack）处理，但如果这么做，缺点是包的往返时间越长性能就越差

为了解决此问题，引入了窗口概念，窗口大小即决定了无需等待应答而可以继续发送的数据最大值

窗口实际就起到一个缓冲区的作用，同时也能起到流量控制的作用

图中深色的部分即要发送的数据，高亮的部分即窗口

窗口内的数据才允许被发送，当应答未到达前，窗口必须停止滑动

如果 1001~2000 这个段的数据 ack 回来了，窗口就可以向前滑动

接收方也会维护一个窗口，只有落在窗口内的数据才能允许接收

MSS 限制

链路层对一次能够发送的最大数据有限制，这个限制称之为 MTU（maximum transmission unit），不同的链路设备的 MTU 值也有所不同，例如

以太网的 MTU 是 1500

FDDI（光纤分布式数据接口）的 MTU 是 4352

本地回环地址的 MTU 是 65535 - 本地测试不走网卡

MSS 是最大段长度（maximum segment size），它是 MTU 刨去 tcp 头和 ip 头后剩余能够作为数据传输的字节数

ipv4 tcp 头占用 20 bytes，ip 头占用 20 bytes，因此以太网 MSS 的值为 1500 - 40 = 1460

TCP 在传递大量数据时，会按照 MSS 大小将数据进行分割发送

MSS 的值在三次握手时通知对方自己 MSS 的值，然后在两者之间选择一个小值作为 MSS

Nagle 算法

即使发送一个字节，也需要加入 tcp 头和 ip 头，也就是总字节数会使用 41 bytes，非常不经济。因此为了提高网络利用率，tcp 希望尽可能发送足够大的数据，这就是 Nagle 算法产生的缘由

该算法是指发送端即使还有应该发送的数据，但如果这部分数据很少的话，则进行延迟发送
如果 SO_SNDBUF 的数据达到 MSS，则需要发送

如果 SO_SNDBUF 中含有 FIN（表示需要连接关闭）这时将剩余数据发送，再关闭

如果 TCP_NODELAY = true，则需要发送

已发送的数据都收到 ack 时，则需要发送

上述条件不满足，但发生超时（一般为 200ms）则需要发送

除上述情况，延迟发送

四、粘包半包代码

粘包

server端代码

public class HelloWorldServer {static final Logger log = LoggerFactory.getLogger(HelloWorldServer.class);void start() {NioEventLoopGroup boss = new NioEventLoopGroup(1);NioEventLoopGroup worker = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.group(boss, worker);serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {log.debug("connected {}", ctx.channel());super.channelActive(ctx);}@Overridepublic void channelInactive(ChannelHandlerContext ctx) throws Exception {log.debug("disconnect {}", ctx.channel());super.channelInactive(ctx);}});}});ChannelFuture channelFuture = serverBootstrap.bind(8080);log.debug("{} binding...", channelFuture.channel());channelFuture.sync();log.debug("{} bound...", channelFuture.channel());channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("server error", e);} finally {boss.shutdownGracefully();worker.shutdownGracefully();log.debug("stoped");}}public static void main(String[] args) {new HelloWorldServer().start();}
}

client端：希望发送 10 个消息，每个消息是 16 字节

public class HelloWorldClient {static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);public static void main(String[] args) {NioEventLoopGroup worker = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(worker);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {log.debug("connected...");ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {log.debug("sending...");//客户端代码希望发送 10 个消息，每个消息是 16 字节for (int i = 0; i < 10; i++) {ByteBuf buffer = ctx.alloc().buffer();buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});ctx.writeAndFlush(buffer);}}});}});ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("client error", e);} finally {worker.shutdownGracefully();}}
}

可以看到server一次就接收了 160 个字节，而非分 10 次接收。

半包

client ：客户端代码发送 1 个消息，这个消息是 160 字节

public class HelloWorldClient {static final Logger log = LoggerFactory.getLogger(HelloWorldClient.class);public static void main(String[] args) {NioEventLoopGroup worker = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(worker);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {log.debug("connected...");ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {log.debug("sending...");//客户端代码发送 1 个消息，这个消息是 160 字节ByteBuf buffer = ctx.alloc().buffer();for (int i = 0; i < 10; i++) {buffer.writeBytes(new byte[]{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15});}ctx.writeAndFlush(buffer);}});}});ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("client error", e);} finally {worker.shutdownGracefully();}}
}

为现象明显，服务端修改一下接收缓冲区，其它代码不变

//注意:这行影响的底层接收缓冲区（即滑动窗口）大小，仅决定了 netty 读取的最小单位，netty 实际每次读取的一般是它的整数倍
serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);

可以看到接收的消息被分为两节，第一次 20 字节，第二次 140 字节（半包）

五、解决方案

改成短连接 (参考UDP)

将 TCP 连接改成短连接，一个请求一个短连接。这样的话，建立连接到释放连接之间的消息即为传输的信息，消息也就产生了边界。这样的方法就是十分简单，不需要在我们的应用中做过多修改。但缺点也就很明显了，效率低下，TCP连接和断开都会涉及三次握手以及四次握手，每个消息都会涉及这些过程，十分浪费性能。 因此，并不推介这种方式。

封装成帧
封装成帧(Framing)，也就是原本发送消息的单位是缓冲大小，现在换成了帧，这样我们就可以自定义边界了。一般有4种方式：

固定长度这种方式下，消息边界也就是固定长度即可。优点就是实现很简单，缺点就是空间有极大的浪费，如果传递的消息中大部分都比较短，这样就会有很多空间是浪费的。 因此，这种方式一般也是不推介的。

分隔符这种方式下，消息边界也就是分隔符本身。优点是空间不再浪费，实现也比较简单。缺点是当内容本身出现分割符时需要转义，所以无论是发送还是接受，都需要进行整个内容的扫描。因此，这种方式效率也不是很高，但可以尝试使用。

专门的 length 字段这种方式，就有点类似 Http 请求中的 Content-Length，有一个专门的字段存储消息的长度。作为服务端，接受消息时，先解析固定长度的字段（length字段）获取消息总长度，然后读取后续内容。优点是精确定位用户数据，内容也不用转义。缺点是长度理论上有限制，需要提前限制可能的最大长度从而定义长度占用字节数。 因此，十分推介用这种方式。

六、Netty 中的实现

Netty 支持上文所讲的封装成帧(Framing)中的前三种方式，简单介绍下：

方式	解码	编码
固定长度	FixedLengthFrameDecoder	简单（不内置）
分割符	DelimiterBasedFrameDecoder （需指定最大长度和自定义Bytebuf类型的分隔符）	简单（不内置）
专门的 length 字段	LengthFieldBasedFrameDecoder	LengthFieldPrepender

FixedLengthFrameDecoder

/*** @author cristianoxiaoming@gmail.com* @version 1.0* @ClassName MyFixedLengthFrameClient* @Package MyNetty.Netty.StickyUnpackingPacket* @ProjectName javaStudy* @description TODO* @date 2021-09-24 11:04* @ProductName IntelliJ IDEA*/
@Slf4j
public class MyFixedLengthFrameClient {public static void main(String[] args) {send();System.out.println("finish");}public static byte[] fill10Bytes(char c, int len) {byte[] bytes = new byte[10];Arrays.fill(bytes, (byte) '_');for (int i = 0; i < len; i++) {bytes[i] = (byte) c;}System.out.println(new String(bytes));return bytes;}private static void send() {NioEventLoopGroup worker = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(worker);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {// 会在连接 channel 建立成功后，会触发 active 事件@Overridepublic void channelActive(ChannelHandlerContext ctx) {ByteBuf buf = ctx.alloc().buffer();char c = '0';Random r = new Random();for (int i = 0; i < 10; i++) {//发送10次信息，每次长度由1到10个字节随机byte[] bytes = fill10Bytes(c, r.nextInt(10) + 1);c++;buf.writeBytes(bytes);}ctx.writeAndFlush(buf);}});}});ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("client error", e);} finally {worker.shutdownGracefully();}}
}

/*** @author cristianoxiaoming@gmail.com* @version 1.0* @ClassName MyFixedLengthFrameServer* @Package MyNetty.Netty.StickyUnpackingPacket* @ProjectName javaStudy* @description FixedLengthFrameDecoder测试* @date 2021-09-24 11:03* @ProductName IntelliJ IDEA*/
@Slf4j
public class MyFixedLengthFrameServer {void start() {NioEventLoopGroup boss = new NioEventLoopGroup();NioEventLoopGroup worker = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);// 调整系统的接收缓冲区（滑动窗口）
//            serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);// 调整 netty 的接收缓冲区（byteBuf）serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(16, 16, 16));serverBootstrap.group(boss, worker);serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//一定要将FixedLengthFrameDecoder刚在第一个处理（不然拿着没解码的桢进行处理也没用），在执行其他handlerch.pipeline().addLast(new FixedLengthFrameDecoder(10));ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));}});ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("server error", e);} finally {boss.shutdownGracefully();worker.shutdownGracefully();}}public static void main(String[] args) {new MyFixedLengthFrameServer().start();}
}

可以看到，客户端一次性将10条信息100字节一次性发出去，这里已经发生了粘包。

但是可以看到，server端已经根据设置的framLength=10进行分包了，接收的时候是正常的。

DelimiterBasedFrameDecoder

/*** @author cristianoxiaoming@gmail.com* @version 1.0* @ClassName MyDelimiterBasedFrameDecoderServer* @Package MyNetty.Netty.StickyUnpackingPacket* @ProjectName javaStudy* @description TODO* @date 2021-09-24 11:25* @ProductName IntelliJ IDEA*/
@Slf4j
public class MyDelimiterBasedFrameDecoderServer {void start() {NioEventLoopGroup boss = new NioEventLoopGroup();NioEventLoopGroup worker = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);// 调整系统的接收缓冲区（滑动窗口）
//            serverBootstrap.option(ChannelOption.SO_RCVBUF, 10);// 调整 netty 的接收缓冲区（byteBuf）serverBootstrap.childOption(ChannelOption.RCVBUF_ALLOCATOR, new AdaptiveRecvByteBufAllocator(16, 16, 16));serverBootstrap.group(boss, worker);serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {//超过1024个字节，还没有遇到换行符就会失败ch.pipeline().addLast(new LineBasedFrameDecoder(1024));ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));}});ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("server error", e);} finally {boss.shutdownGracefully();worker.shutdownGracefully();}}public static void main(String[] args) {new MyDelimiterBasedFrameDecoderServer().start();}
}

/*** @author cristianoxiaoming@gmail.com* @version 1.0* @ClassName MyDelimiterBasedFrameDecoderClient* @Package MyNetty.Netty.StickyUnpackingPacket* @ProjectName javaStudy* @description TODO* @date 2021-09-24 12:52* @ProductName IntelliJ IDEA*/
@Slf4j
public class MyDelimiterBasedFrameDecoderClient {public static void main(String[] args) {send();System.out.println("finish");}public static StringBuilder makeString(char c, int len) {StringBuilder sb = new StringBuilder(len + 2);for (int i = 0; i < len; i++) {sb.append(c);}sb.append("\n");return sb;}private static void send() {NioEventLoopGroup worker = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);bootstrap.group(worker);bootstrap.handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {// 会在连接 channel 建立成功后，会触发 active 事件@Overridepublic void channelActive(ChannelHandlerContext ctx) {ByteBuf buf = ctx.alloc().buffer();char c = '0';Random r = new Random();for (int i = 0; i < 10; i++) {StringBuilder sb = makeString(c, r.nextInt(256) + 1);c++;buf.writeBytes(sb.toString().getBytes());}ctx.writeAndFlush(buf);}});}});ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8080).sync();channelFuture.channel().closeFuture().sync();} catch (InterruptedException e) {log.error("client error", e);} finally {worker.shutdownGracefully();}}
}

LengthFieldBasedFrameDecoder :基于长度字段的桢解码器

/*** @author cristianoxiaoming@gmail.com* @version 1.0* @ClassName MyLengthFieldDecoder* @Package MyNetty.Netty.StickyUnpackingPacket* @ProjectName javaStudy* @description TODO* @date 2021-09-24 13:09* @ProductName IntelliJ IDEA*/
public class MyLengthFieldDecoder {public static void main(String[] args) {EmbeddedChannel channel = new EmbeddedChannel(new LengthFieldBasedFrameDecoder(//int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip1024, 0, 4, 1,4),new LoggingHandler(LogLevel.DEBUG));//  用4个字节表示内容长度，后再接实际内容ByteBuf buffer = ByteBufAllocator.DEFAULT.buffer();send(buffer, "Hello, world");send(buffer, "Hi!");channel.writeInbound(buffer);}private static void send(ByteBuf buffer, String content) {byte[] bytes = content.getBytes(); // 实际内容int length = bytes.length; // 实际内容长度buffer.writeInt(length);buffer.writeByte(1);//额外字节，就要有lengthAdjustmentbuffer.writeBytes(bytes);}
}

参考视频
参考文章