使用按行切换文本解码器解决TCP粘包问题03

TCP粘包/拆包
TCP是个"流"协议，所谓流，就是没有界限的一串数据。一个完整的包可能会被TCP拆分成多个包进行发送，也有可能把多个小包合成一个大的数据包进行发送。这就是所谓的TCP粘包/拆包问题。
举例：
客户端给服务端发送两个数据包：D1和D2；
1、服务读分两次读取到了两个独立的数据包，分别为D1和D2,没有发生粘包/拆包。
2、服务端一次接收到了两个数据包，D1和D2粘合在一起，被称为TCP粘包。
3、服务端分两次读取到了两个数据包，第一次读取到了完整的D1包和D2包的部分内容，第二次读取到了D2剩余部分，这称为TCP拆包。
--------
粘包/拆包发生的原因
1、应用程序write写入的字节大小大于套接口发送缓冲大小
2、进行MSS大小的TCP分段
3、以太网帧payload大于MTU进行IP分片
---
应用程序——TCP——IP——输出队列/数据链路
--------
粘包问题的解决策略：
1、消息定长，例如每个报文的大小固定长度200字节，如果不够，空位补空格；
2、在包尾增加回车换行符进行分割，例如FTP协议
3、将消息分为消息头和消息体，消息头中包含表示消息总长度的字段，通常设计思路为消息头的第一个字段使用int32来表示的总长度；
4、更复杂的应用层协议；

-------------------

使用按行切换文本解码器

LineBasedFrameDecoder与StringDecoder原理
LineBasedFrameDecoder的工作原理是它依次遍历ByteBuf中可读字节，判断是否有"\n"或者"\r\n"，如果有就以此位置结束位置，从可读索引到结束位置区间的字节就组成一行。
StringDecoder将接收到的对象转换成字符串，然后继续调用后面的handler。
LineBasedFrameDecoder与StringDecoder组合就是按行切换文本解码器。

------------
TimeServer:netty服务端

public class TimeServer {

    public void bind(int port) throws Exception{
        //配置服务端的NIO线程组,ChildChannelHandler专门用于处理网络事件
        //一个用于服务端接受客户端的连接，一个用于进行socketChannel的网络读写
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //用于启动NIO服务端的辅助类
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup,workerGroup)
                .channel(NioServerSocketChannel.class)//功能是对应与JDKNIO中的ServerSocketChannel
                .option(ChannelOption.SO_BACKLOG, 1024)//设置NioServerSocketChannel的TCP参数
                .childHandler(new ChildChannelHandler());//绑定IO事件处理类，主要用于处理网络IO事件，例如记录日志、对消息进行编码等
            //绑定端口，同步等待成功，sync()同步阻塞方法
            ChannelFuture f = b.bind(port).sync();
            //等待服务端监听端口关闭
            f.channel().closeFuture().sync();
        } finally {
            //释放线程池资源
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{

        @Override
        protected void initChannel(SocketChannel arg0) throws Exception {
            //解码器，利用 LineBasedFrameDecoder解决TCP半包问题
            arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
            arg0.pipeline().addLast(new StringDecoder());
            //处理类
            arg0.pipeline().addLast(new TimeServerHandler());
        }
        
    }
    
    public static void main(String[] args) throws Exception {
        int port = 8088;
        new TimeServer().bind(port);
    }    
}

服务端处理类

public class TimeServerHandler extends ChannelHandlerAdapter{

    private int counter;
    
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        //在TimeServer加入了StringDecoder()解码器，直接将msg转换为String
        String body = (String) msg;
        System.out.println("the time server receive order:" + body + 
                " ; the couter is: "+ (++counter));//每读到一条消息，计数器加1
        //如果消息是QUERY TIME ORDER则创建应答消息，发送给客户端
        String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body)?new 
                java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
        currentTime = currentTime + System.getProperty("line.separator");
        //将消息放入缓冲区中,再调用flush方法发送，
        ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
        ctx.writeAndFlush(resp);
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        ctx.flush();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //发送异常，关闭ChannelHandlerContext
        ctx.close();
    }    
}

客户端

public class TimeClient {

    public void connect (int port,String host) throws Exception{
        //配置客户端NIO线程组
        EventLoopGroup group =  new NioEventLoopGroup();
        try {
            //启动客户端辅助类
            Bootstrap b = new Bootstrap();
            b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, true)//Tcop参数
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override//匿名内部类，处理IO网络事件TimeClientHandler
                    protected void initChannel(SocketChannel ch) throws Exception {
                        //解码器
                        ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                        ch.pipeline().addLast(new StringDecoder());
                        ch.pipeline().addLast(new TimeClientHandler());
                    }
                });
            //发起异步连接操作
            ChannelFuture f = b.connect(host,port).sync();
            //等待客户端链路关闭
            f.channel().closeFuture().sync();
        }finally{
            group.shutdownGracefully();
        }
    }
    
    public static void main(String[] args){
        int port = 8088;
        try {
            new TimeClient().connect(port, "127.0.0.1");
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
}

客户端处理类

public class TimeClientHandler extends ChannelHandlerAdapter{
    private byte[] req;
    private int counter;
    
    public TimeClientHandler(){
        req = ("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //循环发送100条信息，发生粘包问题
        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("Now is:" + body + " ; the counter is :" + (++counter));
    }
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        //释放资源
        ctx.close();
    }        
}