Netty粘包拆包问题说明、演示拆包粘包情况代码以及解决

概述

TCP传输协议是面向连接的，面向流提供高可靠的服务。收发两端（服务端和客户端）都要有一一成对的socket，因此，发送端为了将多个发给接收端的包，更有效地发给对方，使用了优化算法（Nagle算法），将多次间隔时间较小且数据较小的数据包，合成一个大的数据块，然后进行封包，这样做虽然提高了传输的效率，但是这样接收端就难以分辨出一个个完整的包的大小了，因为面向流的通信时无消息保护边界的。

由于TCP无消息保护边界，需要在接收端处理消息边界问题，也就是粘包、拆包问题。

示意图：

说明：

服务端给客户端发送D1、D2两个数据包。
第一种情况是服务端两次都读取到的是两个独立的数据包，无拆包粘包问题，可以正常解析。
第二种情况D1、D2包合并在一起一次发送，服务端一次性接收了两个数据包，分别是D1、D2，我们称之为粘包，因为不知道边界，所以服务端不知如何拆出来解析。
第三种情况，服务端第一次接收到了完整D1包和部分D2包，第二次接收到了剩余的D2包，D2包被拆开来分多次发送了，我们称之为拆包。
第四种情况，服务端第一次接收到了部分D1包，第二次接收到了剩余部分D1包和完整D2包，也称之为拆包。

总的来说拆包和粘包问题，因为没有边界，最终会在正常情况下导致接收方无法分辨出一个一个包。

粘包实例

以下实例证实了粘包的存在：
服务端及服务端的Handler：

public class NettyTcpServer {private  int port;public NettyTcpServer(int port){this.port = port;}public void start() throws InterruptedException {NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new NettyTcpServerChannelHandler());}});ChannelFuture channelFuture = serverBootstrap.bind(port);channelFuture.channel().closeFuture().sync();}public static void main(String[] args) throws InterruptedException {NettyTcpServer nettyTcpServer = new NettyTcpServer(8989);nettyTcpServer.start();}
}public class NettyTcpServerChannelHandler extends ChannelInboundHandlerAdapter {private int count;@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;System.out.println("数据是：" + byteBuf.toString(CharsetUtil.UTF_8));System.out.println("count = " + (++count));super.channelRead(ctx, msg);}
}

客户端及客户端的Handler

public class NettyTcpClient {private  int serverPort;public NettyTcpClient(int serverPort){this.serverPort = serverPort;}public void start() throws InterruptedException {NioEventLoopGroup worker = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();bootstrap.group(worker).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new NettyTcpClientChannelHandler());}});ChannelFuture connect = bootstrap.connect("127.0.0.1", serverPort);connect.channel().closeFuture().sync();}public static void main(String[] args) throws InterruptedException {NettyTcpClient nettyTcpClient = new NettyTcpClient(8989);nettyTcpClient.start();}
}public class NettyTcpClientChannelHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//发生这个事件代表通道已经顺利连接到远端，可以收发数据。我们就在这里发送数据、//分十次发送for (int i = 0;i<10;i++){ctx.channel().writeAndFlush(Unpooled.copiedBuffer("Hello I am John;", CharsetUtil.UTF_8));}ctx.fireChannelActive();}
}

执行结果：

这个是服务端的输出结果，如果是没有拆包粘包的话，客户端分十次发送了数据，那么服务端也应该分十次接收，count应该等于10，但是此时count=1，然后数据一次性显示出来了，说明客户端的数据时把十次合并成一个数据包发送的，数据粘包现象。

拆包演示

将一次发送的数据量弄大一点，比mss大，就会发生拆包现象。
上面粘包演示的客户端和服务端不变，把两个Handler修改一下。

@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//发生这个事件代表通道已经顺利连接到远端，可以收发数据。我们就在这里发送数据、//分十次发送//一次发送102400字节数据byte[] bytes = new byte[102400];Arrays.fill(bytes, (byte) 10);for (int i = 0;i<10;i++){ctx.channel().writeAndFlush(Unpooled.copiedBuffer(bytes));}ctx.fireChannelActive();}@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = (ByteBuf) msg;System.out.println("长度是：" + byteBuf.readableBytes());System.out.println("count = " + (++count));super.channelRead(ctx, msg);
}

执行结果：

每次发送的数据长度都小于102400，并且发送次数count>10次，所以发生了拆包。

解决方案

使用自定义协议加编码解码器来解决。
需要对Netty的解码编码机制有一定的理解。

关键就是定义每一个数据包的边界，让服务端知道如何对粘包进行拆分和对拆包进行拆分和组合，从而避免拆包粘包问题。

服务端代码

public class NettyTcpServer {private  int port;public NettyTcpServer(int port){this.port = port;}public void start() throws InterruptedException {NioEventLoopGroup bossGroup = new NioEventLoopGroup();NioEventLoopGroup workerGroup = new NioEventLoopGroup();ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();//这里添加一个解码器，顺序必须在业务处理的Handler前面。//因为要对数据先解码了才能进行业务处理。pipeline.addLast(new MessageProtocolDecoder());pipeline.addLast(new NettyTcpServerChannelHandler());}});ChannelFuture channelFuture = serverBootstrap.bind(port);channelFuture.channel().closeFuture().sync();}public static void main(String[] args) throws InterruptedException {NettyTcpServer nettyTcpServer = new NettyTcpServer(8989);nettyTcpServer.start();}
}

客户端：

public class NettyTcpClient {private  int serverPort;public NettyTcpClient(int serverPort){this.serverPort = serverPort;}public void start() throws InterruptedException {NioEventLoopGroup worker = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();bootstrap.group(worker).channel(NioSocketChannel.class).handler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ChannelPipeline pipeline = ch.pipeline();pipeline.addLast(new NettyTcpClientChannelHandler());//这里添加一个编码器，对发送的数据进行编码，因为编码器属于出站的Handler。//而上面的业务逻辑Handler：NettyTcpClientChannelHandler属于入站Handler，所以添加顺序在这里暂时还没所谓。pipeline.addLast(new MessageProtocolEncode());}});ChannelFuture connect = bootstrap.connect("127.0.0.1", serverPort);connect.channel().closeFuture().sync();}public static void main(String[] args) throws InterruptedException {NettyTcpClient nettyTcpClient = new NettyTcpClient(8989);nettyTcpClient.start();}
}

协议类：

@Data
public class MessageProtocol {//内容private byte[] content;//长度private int len;public MessageProtocol(byte[] content){this.content = content;//计算出长度并赋值this.len = content.length;}
}

服务端的业务处理Handler：

public class NettyTcpServerChannelHandler extends ChannelInboundHandlerAdapter {private int count;@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {MessageProtocol messageProtocol = (MessageProtocol) msg;System.out.println("数据是：" + new String(messageProtocol.getContent(),CharsetUtil.UTF_8) + ",长度是：" + messageProtocol.getLen());System.out.println("count = " + (++count));super.channelRead(ctx, msg);}
}

客户端的业务处理类：


//由原来的直接发送字符串，修改成现在使用协议对象来发送
public class NettyTcpClientChannelHandler extends ChannelInboundHandlerAdapter {@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//发生这个事件代表通道已经顺利连接到远端，可以收发数据。我们就在这里发送数据、//分十次发送for (int i = 0;i<10;i++){byte[] content = "Hello I am John;".getBytes(CharsetUtil.UTF_8);//创建协议对象MessageProtocol message = new MessageProtocol(content);//发送ctx.channel().writeAndFlush(message);}super.channelActive(ctx);}
}

服务端的解码器：

public class MessageProtocolDecoder extends ByteToMessageDecoder {@Overrideprotected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {//记录初始的读偏移量，用于拆包情况下的偏移量回滚int beginReadIndex = in.readerIndex();//首先读数据的长度出来先int len = in.readInt();if (in.readableBytes() < len){//这种情况下就是发生了拆包问题，因读取的数据长度没有len长，这个包还有数据没有到达。//重置读偏移量,等待下次重新读写in.readerIndex(beginReadIndex);//直接返回，等待下次return;}//能到这里，就证明包是完整的//读数据byte[] data = new byte[len];in.readBytes(data);MessageProtocol messageProtocol = new MessageProtocol(data);//写出给下一个Handler处理。out.add(messageProtocol);}
}

客户端的编码器：

public class MessageProtocolEncode  extends MessageToByteEncoder<MessageProtocol> {@Overrideprotected void encode(ChannelHandlerContext ctx, MessageProtocol msg, ByteBuf out) throws Exception {//对协议的编码时，把数据的前4个字节固定为数据的长度，跟着就是数据的内容int len = msg.getLen();//先写数据的长度out.writeInt(len);//再写数据的内容out.writeBytes(msg.getContent());}
}

结果：

由上图可见，解决了粘包问题。

下面来测试是否解决了拆包问题。发送的数据使用上面的拆包演示代码，修改一下成用协议即可。
修改后的业务逻辑处理Handler

//客户端@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {//发生这个事件代表通道已经顺利连接到远端，可以收发数据。我们就在这里发送数据、//分十次发送//一次发送102400字节数据byte[] bytes = new byte[102400];Arrays.fill(bytes, (byte) 10);for (int i = 0;i<10;i++){MessageProtocol messageProtocol = new MessageProtocol(bytes);ctx.channel().writeAndFlush(messageProtocol);}ctx.fireChannelActive();}
//服务端
@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {MessageProtocol messageProtocol = (MessageProtocol) msg;System.out.println("长度是：" + messageProtocol.getLen());System.out.println("count = " + (++count));super.channelRead(ctx, msg);}

运行结果：

由上图可知拆包问题已经解决，每次读取的数据都为102400.这里就只打印长度了，因为数据太大，就不打印了。

至此，完毕。