Netty权威指南之TCP粘包和拆包

本章节需要学习的内容如下：

1、TCP粘包和拆包基础知识。

2、TCP粘包和拆包异常案例

3、Netty 解决半包问题。

第一节:TCP粘包和拆包基础知识

TCP粘包和拆包问题产生原因？

TCP协议本质上就是个“流”协议，就是没有界限的一串数据。大家可以想象河里的水，是连成一片，期间没有分界线。TCP底层并不了解上层业务数据的具体含义，它会根据TCP缓冲区的实际情况进行包的划分，所以在业务上认为，一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把大大小小的数据包封装成一个大的数据包进行发送，这就是所谓的TCP粘包和拆包问题。

TCP 粘包和拆包通信模型图：

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

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

2、服务端一次接受两个数据包，D1和D2粘合在一起，称之为粘包。

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

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

如果此时服务端TCP接受划窗非常小，而数据包D1和D2都比较大，很可能发送第五种情况：及服务端多次才能将D1和D2接受完全，期间发生多次拆包。

TCP粘包和拆包发送的原因：

问题产生的原因有三个，分别如下：

1、应用程序write写入的字节大小大于套接字发送缓存大小

2、进行MSS大小的TCP分段

3、以太网帧的playload大于MTU进行IP分片。

粘包问题解决策略:

由于底层TCP无法理解上层的业务数据，所以底层是无法保证数据包不被拆分和重组的，这个问题只能通过上层的应用协议栈设计来解决，根据业界主流协议的解决方案，可以归类如下：

1、消息定长，列如:每个报文的大小固定长度为200个字节，如果不够，空位补空格。

2、在包尾增加回车换行符进行分割，列如：FTP协议

3、将消息分为消息头和消息体，消息头中包含表示消息的总长度（或消息体长度）的字段，通常的设计思路为消息头的第一个字段用int32来表示消息的总长度。

4、更复杂的应用协议。

介绍完TCP的粘包和拆包的基础知识，下面我们通过实践教程看看如何使用Netty提供的半包解码器来解决TCP粘包和拆包问题。

第二节:TCP 粘包和拆包异常案例

在前面的NettyServer服务器代码中，我们多次强调没有考虑半包问题，在功能测试时往往没有什么问题，但是一旦压力上来，或者发送大报文之后，就存在粘包和拆包的问题。如果代码没有考虑，往往会出现解码错位会错误，导致程序不能正常的工作。我们已下面的代码为例，模拟故障现场，然后看看如何正确使用Netty半包解码器，来解决TCP的粘包和拆包问题。

Netty 服务端代码：

NettyServer.java

package com.nio.server; 
 
import com.nio.handler.NettyServerHandler; 
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; 
 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyServer { 
    public static void main(String[] args)throws Exception{ 
        int port=17666; 
        new NettyServer().bind(port); 
 
 
    } 
    public void bind(int port)throws Exception{ 
        //配置服务端的NIO线程池 
        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).childHandler(new NettyServerHandler()); 
            //绑定端口，等待同步成功 
            ChannelFuture f=b.bind(port).sync(); 
            //等待服务端关闭监听端口 
            f.channel().closeFuture().sync(); 
 
        }finally { 
            //释放线程池资源 
            bossGroup.shutdownGracefully(); 
            workGroup.shutdownGracefully(); 
 
        } 
 
 
    } 
    //添加内部类 
    private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{ 
        @Override 
        protected void initChannel(SocketChannel socketChannel) throws Exception { 
            socketChannel.pipeline().addLast(new NettyServerHandler()); 
 
        } 
    } 
 
} 

NettyServerHandler.java

package com.nio.handler; 
 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerAdapter; 
import io.netty.channel.ChannelHandlerContext; 
 
 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyServerHandler extends ChannelHandlerAdapter { 
    private int counter; 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close(); 
    } 
 
    @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 server receiver data is:"+body+"; the counter is:"+ ++counter); 
        String currentTime="time".equalsIgnoreCase(body) ? new java.util.Date(System.currentTimeMillis()).toString():"no zuo no die"; 
        currentTime=currentTime+ System.getProperty("line.separator"); 
        ByteBuf resp= Unpooled.copiedBuffer(currentTime.getBytes()); 
       ctx.writeAndFlush(resp); 
    } 
 
 
} 

Netty客户端代码：

NettyClient.java

package com.nio.client; 
 
import com.nio.handler.NettyClientHandler; 
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; 
 
 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyClient { 
    public static void main(String[] args)throws  Exception{ 
        int port=17666; 
        new NettyClient().connect(port,"127.0.0.1"); 
 
    } 
 
    public void connect(int port,String host)throws Exception{ 
        //配置客户端NIO线程池 
        EventLoopGroup workGroup=new NioEventLoopGroup(); 
        try{ 
            io.netty.bootstrap.Bootstrap b=new io.netty.bootstrap.Bootstrap(); 
            b.group(workGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).handler(new ChannelInitializer<SocketChannel>() { 
                @Override 
                protected void initChannel(SocketChannel socketChannel) throws Exception { 
                    socketChannel.pipeline().addLast(new NettyClientHandler()); 
                } 
            }); 
            //发起异步连接操作 
            ChannelFuture f=b.connect(host,port).sync(); 
            //等待客户端链路关闭 
            f.channel().closeFuture().sync(); 
 
        }finally { 
            //释放NIO 线程组 
            workGroup.shutdownGracefully(); 
 
        } 
    } 
 
 
} 

NettyClientHandler.java

package com.nio.handler; 
 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerAdapter; 
import io.netty.channel.ChannelHandlerContext; 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyClientHandler extends ChannelHandlerAdapter { 
    private int counter; 
 
    private byte[] req; 
 
    public NettyClientHandler() { 
          req=("time"+System.getProperty("line.separator")).getBytes(); 
 
    } 
 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
     ctx.close(); 
 
    } 
 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        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 { 
 
        ByteBuf buf=(ByteBuf)msg; 
        byte[] req=new byte[buf.readableBytes()]; 
        buf.readBytes(req); 
        String body=new String(req,"UTF-8"); 
        System.out.println("this client receiver data is:"+body+"; this is counter is"+ ++counter); 
    } 
 
    @Override 
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
 
    } 
} 

NettyServer 服务端执行效果截图:

按照设计初衷，服务端应该接收到100条消息，但实际上服务端只接收到了一条请求信息。这说明发生了TCP粘包情况。

NettyClient 客户端执行效果截图:

按照设计初衷，客户端应该接收到100条当前系统时间，但实际上只收到一条数据信息。这个不难理解，客户端发送的请求只有一次，所以服务器应答也只有一次。

由于上面的程序没有考虑到TCP 粘包和拆包的问题，所以导致我们的程序无法正常运行。

第三节:Netty解决半包问题

为了解决TCP的粘包和拆包导致的半包读写问题，Netty 提供多种编解码器用于处理半包，只有熟练使用这些类库，TCP的粘包和拆包就变得很容易哒。你甚至不需要关心它们，这也是其他框架和Java NIO 原生框架无法匹敌的。

Netty服务端代码

NettuServer.java

package com.nio.server; 
 
import com.nio.handler.NettyServerHandler; 
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; 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyServer { 
    public static void main(String[] args)throws Exception{ 
        int port=17666; 
        new NettyServer().bind(port); 
 
 
    } 
    public void bind(int port)throws Exception{ 
        //配置服务端的NIO线程池 
        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).childHandler(new ChildChannelHandler()); 
            //绑定端口，等待同步成功 
            ChannelFuture f=b.bind(port).sync(); 
            //等待服务端关闭监听端口 
            f.channel().closeFuture().sync(); 
 
        }finally { 
            //释放线程池资源 
            bossGroup.shutdownGracefully(); 
            workGroup.shutdownGracefully(); 
 
        } 
 
 
    } 
    //添加内部类 
    private class ChildChannelHandler extends 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 NettyServerHandler()); 
 
        } 
    } 
 
} 

NettyServerHandler.java

package com.nio.handler; 
 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerAdapter; 
import io.netty.channel.ChannelHandlerContext; 
 
 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyServerHandler extends ChannelHandlerAdapter { 
    private int counter; 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close(); 
    } 
 
    @Override 
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
 
        String body=(String)msg; 
        System.out.println("the server receiver data is:"+body+"; the counter is:"+ ++counter); 
        String currentTime="time".equalsIgnoreCase(body) ? new java.util.Date(System.currentTimeMillis()).toString():"no zuo no die"; 
        currentTime=currentTime+ System.getProperty("line.separator"); 
        ByteBuf resp= Unpooled.copiedBuffer(currentTime.getBytes()); 
       ctx.writeAndFlush(resp); 
    } 
 
 
} 

Netty客户端代码

NettyClient.java

package com.nio.client; 
 
import com.nio.handler.NettyClientHandler; 
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 io.netty.handler.codec.LineBasedFrameDecoder; 
import io.netty.handler.codec.string.StringDecoder; 
 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyClient { 
    public static void main(String[] args)throws  Exception{ 
        int port=17666; 
        new NettyClient().connect(port,"127.0.0.1"); 
 
    } 
 
    public void connect(int port,String host)throws Exception{ 
        //配置客户端NIO线程池 
        EventLoopGroup workGroup=new NioEventLoopGroup(); 
        try{ 
            io.netty.bootstrap.Bootstrap b=new io.netty.bootstrap.Bootstrap(); 
            b.group(workGroup).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY,true).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 NettyClientHandler()); 
                } 
            }); 
            //发起异步连接操作 
            ChannelFuture f=b.connect(host,port).sync(); 
            //等待客户端链路关闭 
            f.channel().closeFuture().sync(); 
 
        }finally { 
            //释放NIO 线程组 
            workGroup.shutdownGracefully(); 
 
        } 
    } 
 
 
} 

NettyClientHandler.java

package com.nio.handler; 
 
import io.netty.buffer.ByteBuf; 
import io.netty.buffer.Unpooled; 
import io.netty.channel.ChannelHandlerAdapter; 
import io.netty.channel.ChannelHandlerContext; 
 
/** 
 * Created by vixuan-008 on 2015/6/19. 
 */ 
public class NettyClientHandler extends ChannelHandlerAdapter { 
    private int counter; 
 
    private byte[] req; 
 
    public NettyClientHandler() { 
          req=("time"+System.getProperty("line.separator")).getBytes(); 
 
    } 
 
    @Override 
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
     ctx.close(); 
 
    } 
 
    @Override 
    public void channelActive(ChannelHandlerContext ctx) throws Exception { 
        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 { 
 
 
        String body=(String)msg; 
        System.out.println("this client receiver data is:"+body+"; this is counter is"+ ++counter); 
    } 
 
    @Override 
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
 
    } 
} 

NettyServer 服务端运行截图：

NettyClient 客户端运行截图:

程序的运行结果，完全符合预期。说明通过使用LineBaseFrameDecode和StringDecoder成功解决TCP粘包和拆包导致的半包问题。对于使用者来说，只有将支持半包解的handler添加到ChannlePipline中即可，不需要编写而外的代码，用户使用起来很简单。