粘包拆包

概述

TCP是面向流的协议,TCP在网络上传输的数据就是一连串的数据,完全没有分界线。
TCP协议的底层并不了解上层业务的具体定义,它会根据TCP缓冲区的实际情况进行包的划分。
在业务层面认为一个完整的包可能会被TCP拆分成多个小包进行发送,也可能把多个小的包封装成一个大的数据包进行发送,这就是所谓的TCP粘包拆包问题，如上图会存在多种情况

原因分析 

TCP数据流最终发到目的地,必须通过以太网协议封装成一个个的以太网帧发送出去,以太网数据帧大小最小64字节,最大1518字节,除去header部分,其数据payload为46到1500字节。所以如果以太网帧的payload大于MTU(默认1500字节)就需要进行拆包

解决方案

由于TCP协议底层无法理解上层的业务数据，所以在底层是无法保证数据包不被拆分和重组的，所以该问题只能通过上层的应用层协议设计来解决,常见方案如下：

• 消息定长,发送方和接收方规定固定大小的消息长度,例如每个报文大小固定为200字节，如果不够，空位补空格
• 增加特殊字符进行分割,例如FTP协议
• 自定义协议,将消息分为消息头和消息体，消息头中包含消息总长度，这样服务端就可以知道每个数据包的具体长度了，知道了发送数据包的具体边界后，就可以解决粘包和拆包问题了

Netty解决方案

基于JDK原生的socket或者nio解决粘包拆包问题比较麻烦;作为一款非常强大的网络通信框架,Netty提供了多种编码器用于解决粘包拆包问题,只要掌握这些类库的使用即可解决粘包拆包问题,常见的类库包括：

• LineBasedFrameDecoder：基于行的解码器,遇到 “\n”、"\r\n"会被作为行分隔符
• FixedLengthFrameDecoder：基于固定长度的解码器
• DelimiterBasedFrameDecoder：基于分隔符的帧解码器
• ByteToMessageDecoder：处理自定义消息协议的解码器
  • ReplayingDecoder：继承ByteToMessageDecoder,网络缓慢并且消息格式复杂时可能慢于ByteToMessageDecoder

自定义消息协议+编码/解码器解决粘包拆包问题实战

Netty中使用自定义协议(本质就是解决服务器端每次读取数据长度的问题) + 编码解码器来解决

客户端

package com.bierce.io.netty.stickingAndunpacking;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;import java.nio.charset.Charset;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
public class MyStickingUnpackingClientTest {private String host;private int port;public MyStickingUnpackingClientTest(String host, int port){this.host = host;this.port = port;}public void run(){NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();try {Bootstrap bootstrap = new Bootstrap();bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageEncoder()); //自定义编码器pipeline.addLast(new MyMessageDecoder()); //自定义解码器pipeline.addLast(new MyStickingUnpackingClientHandler()); //自定义业务处理器}});ChannelFuture cf = bootstrap.connect(host, port).sync();Channel channel = cf.channel();System.out.println("北京时间-" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss")) + ": " + "Client"+ channel.localAddress()+ " start Successful!!!");cf.channel().closeFuture().sync();} catch (InterruptedException e) {throw new RuntimeException(e);} finally {eventLoopGroup.shutdownGracefully();}}public static void main(String[] args) {new MyStickingUnpackingClientTest("127.0.0.1", 9999).run();}
}
class MyStickingUnpackingClientHandler extends SimpleChannelInboundHandler<MessageProtocal> {private int count;@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {for (int i = 0; i < 5; i++) {String msg = "Hello Bierce";byte[] data = msg.getBytes(Charset.forName("UTF-8"));int length = msg.getBytes(Charset.forName("UTF-8")).length;MessageProtocal msgProtocal = new MessageProtocal();msgProtocal.setData(data);msgProtocal.setLength(length);ctx.writeAndFlush(msgProtocal);}}@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {byte[] data = msg.getData();int length = msg.getLength();System.out.println("客户端接收到服务端响应信息包括：");System.out.println("长度=" + length + ";内容=" + new String(data, Charset.forName("UTF-8"))  + ";消息包数量=" + (++this.count));}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息： " + cause.getMessage());ctx.close();}
}

服务端

package com.bierce.io.netty.stickingAndunpacking;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import java.nio.charset.Charset;
import java.util.UUID;
public class MyStickingUnpackingServerTest {private int port;public MyStickingUnpackingServerTest(int port){this.port = port;}public void run(){EventLoopGroup bossGroup = new NioEventLoopGroup();EventLoopGroup workerGroup = new NioEventLoopGroup();try {ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).handler(new LoggingHandler(LogLevel.INFO)).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new MyMessageDecoder()); //自定义解码器pipeline.addLast(new MyMessageEncoder()); //自定义编码器pipeline.addLast(new MyStickingUnpackingServerHandler());}});System.out.println("服务器启动成功!");serverBootstrap.bind(port).sync().channel().closeFuture().sync();} catch (InterruptedException e) {throw new RuntimeException(e);} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}public static void main(String[] args) {new MyStickingUnpackingServerTest(9999).run();}
}
class MyStickingUnpackingServerHandler extends SimpleChannelInboundHandler<MessageProtocal> {private int count;@Overrideprotected void channelRead0(ChannelHandlerContext ctx, MessageProtocal msg) throws Exception {int length = msg.getLength();byte[] data = msg.getData();System.out.println();System.out.println("服务器接收到的信息包括：");System.out.println("长度=" + length + ";内容=" + new String(data, Charset.forName("UTF-8"))  + ";消息包数量=" + (++this.count));System.out.println("------------服务端读取成功------------");//回复给客户端信息String responseContent = UUID.randomUUID().toString();int responseLen = responseContent.getBytes("UTF-8").length;MessageProtocal msgProtocal = new MessageProtocal();msgProtocal.setLength(responseLen);msgProtocal.setData(responseContent.getBytes("UTF-8"));ctx.writeAndFlush(msgProtocal);}@Overridepublic void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {System.out.println("异常消息：" + cause.getMessage());ctx.close();}
}

自定义消息协议类

package com.bierce.io.netty.stickingAndunpacking;
/*** 自定义消息服协议*/
public class MessageProtocal {private int length;private byte[] data;public int getLength() {return length;}public void setLength(int length) {this.length = length;}public byte[] getData() {return data;}public void setData(byte[] data) {this.data = data;}
}

自定义消息解码编码器类

package com.bierce.io.netty.stickingAndunpacking;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import io.netty.handler.codec.ReplayingDecoder;
import java.util.List;
public class MyMessageDecoder extends ReplayingDecoder<Void> {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {System.out.println("MyMessageDecoder method called");int length = in.readInt();byte[] data = new byte[length];in.readBytes(data);MessageProtocal messageProtocal = new MessageProtocal();messageProtocal.setLength(length);messageProtocal.setData(data);out.add(messageProtocal);}
}
class MyMessageEncoder extends MessageToByteEncoder<MessageProtocal> {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocal msg, ByteBuf out) throws Exception {System.out.println("MyMessageEncoder method called");out.writeInt(msg.getLength());out.writeBytes(msg.getData());}
}

效果图