netty权威指南学习笔记三——TCP粘包/拆包之粘包现象

　　TCP是个流协议，流没有一定界限。TCP底层不了解业务，他会根据TCP缓冲区的实际情况进行包划分，在业务上，一个业务完整的包，可能会被TCP底层拆分为多个包进行发送，也可能多个小包组合成一个大的数据包进行发送，这就是TCP的拆包和粘包。

　　产生问题的原因

• 　　应用程序write写入的字节大小大于套接字缓冲区大小
• 　　进行MSS大小的TCP分割；
• 　　以太网帧的payload大于MTU进行IP分片

关于MSS和MTU的分段问题可以参考https://www.cnblogs.com/yuyutianxia/p/8120741.html学习。

　　解决策略

　　由于底层的TCP无法理解上层的业务数据，所以在底层无法保证数据包不被拆分和重组，要解决这个问题就需要应用层进行人为干预，主要方案有：

• 　　消息定长，例如每个报文大小为固定长度200字节，如果不够，空位补空格；
• 　　在包尾增加回车换行符进行分割，例如FTP协议；
• 　　将消息分为消息头和消息体，消息头中包含表示消息总长度（或者消息体长度）的字段，通常涉及思路为消息头的第一个字段使用int32来表示消息的总长度；
• 　　其他更复杂的应用层协议。

下面模拟粘包的场景，即在服务端去掉分隔符，而客户端发送大量信息过来

TimeServer 服务端代码

package com.StickyUnpack;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;

public class TimeServer {
    public void bind(int port) throws InterruptedException {
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();//创建接收请求的线程组
        NioEventLoopGroup workGroup = new NioEventLoopGroup();//创建处理IO的线程组
        try {
            ServerBootstrap b = new ServerBootstrap();//创建NIO服务启动辅助类
            b.group(bossGroup,workGroup)//将两个线程组传递到NIO服务辅助启动类中
                    .channel(NioServerSocketChannel.class)//通过反射创建1一个severSocketChannel对象
                    .option(ChannelOption.SO_BACKLOG,1024)//BACKLOG用于构造服务端套接字ServerSocket对象，
                    // 标识当服务器请求处理线程全满时，用于临时存放已完成三次握手的请求的队列的最大长度。
                    // 如果未设置或所设置的值小于1，Java将使用默认值50。
                    .childHandler(new ChildHandler());//设置绑定的IO处理类，用来处理网络IO事件
//          绑定端口，同步等待成功，返回ChannelFutrue,用于异步操作通知回调
            ChannelFuture f = b.bind(port).sync();
//          等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        }finally {
//            优雅的退出，释放线程池资源
            bossGroup.shutdownGracefully();
            workGroup.shutdownGracefully();
        }

    }
//    IO处理类的初始化
    private class ChildHandler extends ChannelInitializer {
        @Override
        protected void initChannel(Channel channel) throws Exception {
            channel.pipeline().addLast(new TimeServerHandler());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        int port = 8080;
        try {
            if(args.length>0&&args[0]!=null){
                System.out.println(args[0]);
                port = Integer.valueOf(args[0]);
            }
        } catch (Exception e) {
            port = 8080;
        }
        new TimeServer().bind(port);
    }
}

服务端处理IO代码

package com.StickyUnpack;

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

import java.util.Date;

public class TimeServerHandler extends ChannelInboundHandlerAdapter {
    int count;
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf byteBuf = (ByteBuf) msg;//将请求转为ByteBuf缓冲区
        byte[] req = new byte[byteBuf.readableBytes()];//获取byteBuf的可读字节数
        byteBuf.readBytes(req);//将缓冲区字节数组复制到req数组中
        String body = new String(req,"utf-8")//转换为字符串
                //改造去掉客户端传递过来的换行符号，模拟故障造成粘包问题
             .substring(0,req.length-System.lineSeparator().length());
        System.out.println("the time server receive order:"+body+"the count is:"+ ++count);
//      处理IO内容
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)
                ?new Date(System.currentTimeMillis()).toString():"BAD ORDER";
        currentTime = currentTime+System.getProperty("line.separator");
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());//返回客户端的消息转化为ByteBuf对象
        ctx.write(resp);//将待应答消息放入缓冲数组中
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();//将应答消息写入SocketChannel中
    }

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

客户发送请求端代码

package com.StickyUnpack;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

public class TimeClient {
    public void connect(String host,int port){
        NioEventLoopGroup workGroup = new NioEventLoopGroup();
        Bootstrap b = new Bootstrap();
        try {
        b.group(workGroup).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ClientChildHandler());
        ChannelFuture f = b.connect(host,port).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }finally {
            workGroup.shutdownGracefully();
        }
    }
    public class ClientChildHandler extends ChannelInitializer<SocketChannel>{
        @Override
        protected void initChannel(SocketChannel socketChannel) throws Exception {
            socketChannel.pipeline().addLast(new TimeClientHandler());
        }
    }
    public static void main(String[] args) {
        int port = 8080;
        if (args.length>0&&args!=null){
            try {
                port = Integer.parseInt(args[0]);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
        new TimeClient().connect("127.0.0.1",port);
    }
}

客户端处理IO代码

package com.StickyUnpack;

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 {
    int count;
    private byte[] req;
//    private final ByteBuf firstMessage;
    public TimeClientHandler() {
        /*byte[] req = "QUERY TIME ORDER".getBytes();
        firstMessage = Unpooled.buffer(req.length);
        firstMessage.writeBytes(req);*/
//        改造代码
        req = ("QUERY TIME ORDER"+System.getProperty("line.separator")).getBytes();
    }

    @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 byteBuf = (ByteBuf) msg;
        byte[] req = new byte[byteBuf.readableBytes()];
        byteBuf.readBytes(req);
        String body = new String(req,"utf-8");
        System.out.println("Now is:"+body+";The client count is:"+ ++count);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext chc, Throwable throwable) throws Exception {
        chc.close();
    }
}

运行结果

客户端运行结果

10:38:35.073 [nioEventLoopGroup-2-1] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.linkCapacity: 16
10:38:35.073 [nioEventLoopGroup-2-1] DEBUG io.netty.util.Recycler - -Dio.netty.recycler.ratio: 8
Now is:BAD ORDER
BAD ORDER
;The client count is:1

服务端运行结果

10:38:35.129 [nioEventLoopGroup-3-1] DEBUG io.netty.util.ResourceLeakDetectorFactory - Loaded default ResourceLeakDetector: io.netty.util.ResourceLeakDetector@6751e055
the time server receive order:QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDthe count is:1
the time server receive order:
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDER
QUERY TIME ORDERthe count is:2

从服务端运行结果来看我们期待服务端接收到100条信息，但是运行结果显示只接收到了2条信息，发生了粘包现象，同样客户端也应该接收到100条相应信息，但因为服务端只接收到2条信息，故而只返回两条相应均为BAD ORDER，但是从服务端运行结果看，却只收到1条应答，说明在服务端返回客户端的两条信息也发生了粘包现象。