《Netty权威指南 NIO 入门篇》

深碍√TFBOYSˉ_ 2022-02-25 11:11 234阅读 0赞

### 《Netty NIO 入门篇》 ###

*  Netty简单介绍
 *  搭建Netty入门应用
 *  TCP粘包、拆包问题及解决方案
 *   *  什么是粘包/拆包
     *  TCP粘包/拆包现象案例
     *  利用LineBasedFrameDecoder解码器解决TCP粘包拆包问题
     *  采用分隔符和定长解码器解决TCP粘包和拆包

# Netty简单介绍 #

为什么选择Netty？开发高质量的NIO程序并不是一件简单的事情，出去NIO的复杂性和BUG不谈，作为一个NIO服务器，要能处理网络的闪断、客户端的重复接入、客户端的安全认证、消息的编解码、半包读写情况，如果没有足够的NIO编程经验累积，一个NIO框架的稳定往往需要半年甚至更长的实际。并且从维护性角度而言，NIO采用了异步非阻塞编程模型，而且是一个I/O线性处理多条链路，调试和跟踪非常麻烦，我们无法进行有效的调试和跟踪，定位难度很大。同时NIO编程设计到Reactor模式，你必须对多线程和网络编程非常熟悉，才能编写出高质量的NIO程序。  
  而Netty具有以下优点：1、API使用简单，开发门槛低。2、预置了多种编解码功能。3、定制能力强，可以通过ChannelHandler对通信框架进行灵活的扩展。4、Netty修复了已经发现的NIO的BUG。Netyy在大数据、互联网等众多行业进行了成功的商用，正因这些优点，Netty成为了 Java NIO编程的首选框架。  
  **Netty相比原生的NIO做了哪些改变**？

# 搭建Netty入门应用 #

从官网上 [https://netty.io/downloads.html][https_netty.io_downloads.html] /maven repostory下载压缩包：新建java项目，将all in one中的netty-all .jar拷贝至工程lib下，并添加至构建路径即可。  
服务端时间服务器：

*  创建ServerBootstrap实例来引导绑定和启动服务器
 *  创建NioEventLoopGroup对象来处理事件，如接受新连接、接收数据、写数据等等
 *  指定InetSocketAddress，服务器监听此端口
 *  设置childHandler执行所有的连接请求
 *  都设置完毕了，最后调用ServerBootstrap.bind()

1、先创建两个NioEventLoopGroup线程组(Reactor线程组)，一个NioEventLoopGroup包含了一组NIO线程，一个线程组用于服务端接收客户端的连接，另一个用于进行SocketChannel的网络读写。  
`EventLoopGroup bossGroup = new NioEventLoopGroup();// 用户服务端接收客户端的链接`  
`EventLoopGroup workerGroup = new NioEventLoopGroup();// 用户进行SocketChannel的网络读写`  
2、创建ServerBootstrap对象，用于辅助NIO服务端启动，目的是降低服务端开发的复杂度。  
`ServerBootstrap b = new ServerBootstrap();`  
3、调用ServerBootstrap的group方法，将上面的两个NIO线程组传入该方法，同时要配置NioServerSocketChannel的TCP参数，还要绑定I/O事件的处理类ChildChannelHandler，事件处理类ChildChannelHandler类似Reactor模式中Handler类，处理网络I/O事件，例如记录日志、对消息进行编码等。ChildChannelHandler需要继承ChannelInitializer抽象类，实现initChannel方法。  
`b.group(bossGroup, workerGroup)// 将两个NIO线程组当做入参传递到ServerBootstrap中`  
`.channel(NioServerSocketChannel.class)`  
`//创建NioServerSocketChannel 对应JDK NIO库中的ServerSocketChannel类 , 指定通道类型为NioServerSocketChannel`  
`.option(ChannelOption.SO_BACKLOG, 1024)`  
`//然后配置NioServerSocketChannel的TCP参数,此处将它的backlog设置为1024`

`//最后绑定IO事件的处理类ChildChannelHandler,类似于Reactor模式中的Handler类,主要用于处理网络IO事件,例如记录日志，消息编解码等`  
`.childHandler(new ChildChannelHandler());`  
4、服务器启动辅助类配置完成后，调用它的bind方法绑定监听端口，然后调用它的同步阻塞方法sync等待绑定完成，之后Netty返回一个ChannelFuture，功能类似java.util.concurrent.Future，用于异步操作的通知回调

ChannelFuture f = b.bind(port).sync();

5、使用f.channel().closeFuture().sync()方法进行阻塞，等待服务端链路关闭之后main函数才退出。

f.channel().closeFuture().sync();

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;
    	
    	public class TimeServer { 
    	    public void bind(int port) throws Exception { 
    			// 配置服务端的NIO线程组 ,专门用于网络事件的处理,实际上他们就是Reactor线程组
    			EventLoopGroup bossGroup = new NioEventLoopGroup();// 用户服务端接收客户端的链接
    			EventLoopGroup workerGroup = new NioEventLoopGroup();// 用户进行SocketChannel的网络读写
    			try { 
    				// ServerBootstrap对象实际上是netty用于启动NIO服务端的辅助启动类，目的是降低服务端的开发复杂度
    				ServerBootstrap b = new ServerBootstrap();
    				b.group(bossGroup, workerGroup)// 将两个NIO线程组当做入参传递到ServerBootstrap中
    					.channel(NioServerSocketChannel.class)//创建NioServerSocketChannel 对应JDK NIO库中的ServerSocketChannel类 , 指定通道类型为NioServerSocketChannel
    					.option(ChannelOption.SO_BACKLOG, 1024)//然后配置NioServerSocketChannel的TCP参数,此处将它的backlog设置为1024
    						//最后绑定IO事件的处理类ChildChannelHandler,类似于Reactor模式中的Handler类,主要用于处理网络IO事件,例如记录日志，对消息进行编解码等
    					.childHandler(new ChildChannelHandler());
    				// 绑定端口，同步方法等待绑定操作完成 ,返回ChannelFuture主要用于异步操作的通知回调.
    				ChannelFuture f = b.bind(port).sync();
    	
    				// 等待阻塞,等待服务端链路关闭之后main函数才退出
    				f.channel().closeFuture().sync();
    			} finally { 
    				// 优雅退出，释放线程池资源
    				bossGroup.shutdownGracefully();
    				workerGroup.shutdownGracefully();
    			}
    	    }
    	    // 调用childHandler 来指定连接后调用的ChannelHandler,这个方法传ChannelInitalizer类型的参数
    		// ChannelInitalizer是个抽象类,所以要实现initChannel方法, 这个方法及时用来设置ChannelHandler的
    	
    	    private class ChildChannelHandler extends ChannelInitializer<SocketChannel> { 
    			@Override
    			protected void initChannel(SocketChannel arg0) throws Exception { 
    				arg0.pipeline().addLast(new TimeServerHandler());
    			}
    	
    	    }
    	    /** * @param args * @throws Exception */
    	    public static void main(String[] args) throws Exception { 
    			int port = 30000;
    			if (args != null && args.length > 0) { 
    				try { 
    				port = Integer.valueOf(args[0]);
    				} catch (NumberFormatException e) { 
    				// 采用默认值
    				}
    			}
    			new TimeServer().bind(port);
    	    }
    	}

*  ChannelHandler : IO事件处理类 处理网络IO事件,例如记录日志，对消息进行编解码等…
 *  ChannelHandlerAdapter : IO事件适配器类 ，进行链接、读、写、等事件的处理及扩展  
    TimeServerHandler 实现业务器业务逻辑，继承了ChannelHandlerAdapter ：Netty使用futures和回调概念，它的设计允许你处理不同的事件类型。你的channelhandler必须继承ChannelInboundHandlerAdapter并且重写channelRead方法，这个方法在任何时候都会被调用来接收数据，在这个例子中接收的是字节。

1、首先将接收到的数据做类型转换，将msg消息转换成Netty的ByteBuf对象，即相当于jdk中的java.nio.ByteBuffer对象，通过ByteBuf的readableBytes方法可以获取缓冲区可读的字节数，根据字节数创建byte数组，通过ByteBuf的readBytes方法将缓冲区的字节数组复制到新建的byte数组中，最后通过NewString构造字符串。由于没有加StringDecoder、StringEncoder因此，此处需要类型转换。

ByteBuf buf = (ByteBuf) msg;
    		byte[] req = new byte[buf.readableBytes()];
    		buf.readBytes(req);
    		String body = new String(req, "UTF-8");

2、在读完后，会调用channelReadComplete方法，由于在读方法中有异步发送应答操作，因此channelReadComplete方法中调用ChannelHandlerContext的flush方法，将消息发送队列中的消息写入到SocketChannel中发送给对象，为了防止频繁的换下Selector进行消息发送，Netty的write方法并不直接将消息写入SocketChannel中，调用write方法只是把待发送的消息放到发送缓冲数组中，再次调用flush方法，将缓冲区中的消息全部写入到SocketChannel。

ctx.write(resp);// 异步发送应答消息给客户端
    		// 再通过调用flush方法,将发送缓冲区中的消息全部写到SocketChannel中
    		ctx.flush();

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    import java.util.logging.Logger;
    
    
    public class TimeServerHandler extends ChannelHandlerAdapter { 
    	private static final Logger logger = Logger
    			.getLogger(TimeServerHandler.class.getName());
        @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("The time server receive order : " + body);
    		String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new java.util.Date(
    			System.currentTimeMillis()).toString() : "BAD ORDER";
    		ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
    		ctx.write(resp);// 异步发送应答消息给客户端
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
    		// 将发送缓冲区中的消息全部写到SocketChannel中发送给对方。
    		// 从性能角度考虑，为了防止频繁地唤醒Selector进行消息发送,Netty的write方法并不直接将消息写入SocketChannel中
    		// 调用write方法只是把待发送的消息放到发送缓冲区数中，
    		// 再通过调用flush方法,将发送缓冲区中的消息全部写到SocketChannel中
    		ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
    		logger.warning("Unexpected exception from downstream : "
    				+ cause.getMessage());
    		/** * 当发生异常时,关闭ChannelHandlerContext，释放ChannelHandlerContext相关联的句柄等资源 */
    		ctx.close();
        }
    }

客户端类：

*  创建Bootstrap对象用来引导启动客户端
 *  创建EventLoopGroup对象并设置到Bootstrap中，EventLoopGroup可以理解为是一个线程池，这个线程池用来处理连接、接受数据、发送数据
 *  创建InetSocketAddress并设置到Bootstrap中，InetSocketAddress是指定连接的服务器地址
 *  添加一个ChannelHandler，客户端成功连接服务器后就会被执行
 *  调用Bootstrap.connect()来连接服务器
 *  最后关闭EventLoopGroup来释放资源

1、客户端首先创建I/O读写的NioEventLoopGroup线程组，然后继续创建客户端辅助启动类Bootstrap，随后对其配置，与服务端不同的是，它的Channel需要设置为NioSocketChannel，然后为其添加Handler，此处为了简单直接创建了匿名内部类，实现initChannel方法，作用是当创建NioSocketChannel，成功之后，进行初始化时，将ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件。

EventLoopGroup group = new NioEventLoopGroup();
    Bootstrap b = new Bootstrap();// 客户端辅助启动类,然后对其进行配置
    			b.group(group).channel(NioSocketChannel.class)
    			 .option(ChannelOption.TCP_NODELAY, true)
    					// 作用是当创建NioSocketChannel成功之后,在进行初始化时,
    					// 将ChannelHandler设置到ChannelPipeline中,用于处理网络IO事件
    			 .handler(new ChannelInitializer<SocketChannel>() {
    					@Override
    					public void initChannel(SocketChannel ch)
    						throws Exception {
    						ch.pipeline().addLast(new TimeClientHandler());
    					}
    				});

import io.netty.bootstrap.Bootstrap;
    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;
    
    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)
    					// 作用是当创建NioSocketChannel成功之后,在进行初始化时,
    					// 将ChannelHandler设置到ChannelPipeline中,用于处理网络IO事件
    				.handler(new ChannelInitializer<SocketChannel>() { 
    					@Override
    					public void initChannel(SocketChannel ch)
    						throws Exception { 
    						ch.pipeline().addLast(new TimeClientHandler());
    					}
    				});
    
    			// 发起异步连接操作
    			ChannelFuture f = b.connect(host, port).sync();
    
    			// 当代客户端链路关闭
    			f.channel().closeFuture().sync();
    		} finally { 
    			// 优雅退出，释放NIO线程组
    			group.shutdownGracefully();
    		}
        }
    
        /** * @param args * @throws Exception */
        public static void main(String[] args) throws Exception { 
    		int port = 30000;
    		if (args != null && args.length > 0) { 
    			try { 
    			port = Integer.valueOf(args[0]);
    			} catch (NumberFormatException e) { 
    			// 采用默认值
    			}
    		}
    		new TimeClient().connect(port, "127.0.0.1");
        }

TimeClientHandler重点关注三个方法：channelActive、channelRead和exceptionCaught，当客户单与服务器TCP链路建立成功后，Netty的NIO线程调用channelActive方法，发送查询时间的指令给服务器端，调用ChannelHandlerContext的writeAndFlush方法将请求发送给服务端。当服务端返回应答消息，channelRead方法被调用。

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    import java.util.logging.Logger;
    public class TimeClientHandler extends ChannelHandlerAdapter { 
        private static final Logger logger = Logger
    	    .getLogger(TimeClientHandler.class.getName());
        private final ByteBuf firstMessage;
        /** * Creates a client-side handler. */
        public TimeClientHandler() { 
    		byte[] req = "QUERY TIME ORDER".getBytes();
    		firstMessage = Unpooled.buffer(req.length);
    		firstMessage.writeBytes(req);
        }
        /** * 当客户端和服务端TCP链路简历成功之后,Netty的NIO线路会调用channelActive方法,发送查询时间的指令给服务器 * @param ctx */
        @Override
        public void channelActive(ChannelHandlerContext ctx) { 
    	    ctx.writeAndFlush(firstMessage);// 调用 writeAndFlush 方法将请求消息发送给服务端
        }
        /** * 当服务端返回应答消息时,channelRead 方法被调用 * @param ctx * @param msg * @throws Exception */
        @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("Now is : " + body);
        }
    
        /** * 发生异常时，打印异常日志，释放客户端资源 * @param ctx * @param cause */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { 
    	// 当发生异常时，打印异常日志，释放客户端资源
    	logger.warning("Unexpected exception from downstream : "
    		+ cause.getMessage());
    	ctx.close();
        }
    }

运行结果：  
![在这里插入图片描述][20190402104918490.png]![在这里插入图片描述][20190402104926712.png]

# TCP粘包、拆包问题及解决方案 #

## 什么是粘包/拆包 ##

**一个完整的业务可能会被TCP拆分成多个包进行发送（拆包）， 也有可能把多个小的包封装成一个大的数据包发送（粘包）， 这个就是TCP的拆包和封包问题**。  
             ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70]  
解析上图：

1.  第一种情况，Data1和Data2都分开发送到了Server端，没有产生粘包和拆包的情况。
2.  第二种情况，Data1和Data2数据粘在了一起，打成了一个大的包发送到Server端，这个情况就是粘包。
3.  第三种情况，Data2被分离成Data2\_1和Data2\_2，并且Data2\_1在Data1之前到达了服务端，这种情况就产生了拆包。

由于网络的复杂性，可能数据会被分离成N多个复杂的拆包/粘包的情况，所以在做TCP服务器的时候就需要首先解决拆包/粘包的问题。

## TCP粘包/拆包现象案例 ##

Server改造上例主要体现在：TimeServerHandler上，每接收到一条消息，就记一次数，然后发送应答消息给客户端。按照设计服务端接收的消息总数应该与客户端发送的消息总数相同，  
**TimeServerHandler**：

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class TimeServerHandler extends ChannelHandlerAdapter{ 
        private int counter;
        @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("The time server receive order : "+body
                    +" ; the counter is : " + ++counter);
            String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
            currentTime=currentTime + System.getProperty("line.separator");
    
            ByteBuf resp =Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.writeAndFlush(resp);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ 
            ctx.close();
        }
    }

**TimeServer**：不改造

客户端在上面的例子中主要更改为客户端与服务端链路创建成功后，在TimeClientHandler中循环发送100条消息，每发送一条就刷新一次，保证每条消息偶读会被写入Channel中，按照这个设计，服务器端应该接收到100条查询事件指令的请求消息。  
**TimeClient** ：与上例不变

import io.netty.bootstrap.Bootstrap;
    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.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    public class TimeClient { 
        public void connect(int port,String host) throws Exception{ 
            EventLoopGroup group=new NioEventLoopGroup();
            try{ 
                Bootstrap b=new Bootstrap();
                //Channel需要设置为NioSocketChannel，然后为其添加Handler
                b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>(){ 
                    //为了简单直接创建匿名内部类，实现initChannel方法
                    //其作用是当创建NioSocketChannel成功之后，在进行初始化时，
                    //将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception{ 
                        ch.pipeline().addLast(new TimeClientHandler());
                    }
                });
                //发起异步连接，然后调用同步方法等待连接成功
                ChannelFuture f=b.connect(host,port).sync();
                //当客户端连接关闭之后，客户端主函数退出，退出前释放NIO线程组的资源
                f.channel().closeFuture().sync();
            }finally{ 
    
            }
        }
        public static void main(String[] args) throws Exception { 
            int port=30000;
            new TimeClient().connect(port, "127.0.0.1");
        }
    }

**TimeClientHandler**如下：

import java.util.logging.Logger;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    public class TimeClientHandler extends ChannelHandlerAdapter{ 
        private static final Logger logger=Logger.getLogger(TimeClientHandler.class.getName());
        private int counter;
        private byte[] req;
        public TimeClientHandler(){ 
            req=("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
        }
        /** * 当客户端和服务器TCP链路建立成功后，NIO线程会调用channelActive方法 */
        @Override
        public void channelActive(ChannelHandlerContext ctx){ 
            //发送查询时间的指令给服务端
            ByteBuf message=null;
            //循环发送100条消息，每发送一条就刷新一次
            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("Now is : " + body + " ; the counter is : " + ++counter);
        }
        /** * 当发生异常时 */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ 
            logger.warning("Unexpected exception from downstrea : " + cause.getMessage());
            ctx.close();
        }
    }

运行结果：client实际上发了100条“QUERY TIME ORDER”指令，但是服务端运行结果表明只接收到了两条消息(经管这两条中包含了100次的QUERY TIME ORDER)，所以count只执行了两次。  
![在这里插入图片描述][20190402155847879.png]  
![在这里插入图片描述][20190402155857352.png]  
同时：服务端由于只收到2条请求消息，所以也只反馈了两条BAD ORDER给服务端。但是Client以为是收到一条包含2条“BAD ORDER”的指令消息，所以服务端返回的应答消息也发生了粘包。  
![在这里插入图片描述][20190402160504537.png]

## 利用LineBasedFrameDecoder解码器解决TCP粘包拆包问题 ##

为了解决TCP粘包/拆包导致的半包读写问题，**Netty默认提供了多种编解码器**，只要数列掌握这些类库，TCP粘包问题变得非常容易，这也是其他NIO框架和原生的NIO API无法匹敌的。  
  我们在原来存在粘包的TimeServer中的ChildChannelHandler新增两个解码器：LineBasedFrameDecoder和StringDecoder。**LineBasedFrameDecoder 是依次遍历ByteBuf中的可读字节，判断看是否有\\n 或 \\r\\n，如果有，就以此位置为结束位置，以换行符为结束标志的解码器。它支持携带结束符或者不携带结束符两种解码方式，同时支持配置单行的最大长度。如果连续读取到最大长度后仍然没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。LineBasedFrameDecoder+StringDecoder组合就是按行切换的文本解码器，它被设计用来支持TCP的粘包和拆包**。

private class ChildChannelHandler extends ChannelInitializer<SocketChannel>{
            @Override
            protected void initChannel(SocketChannel arg0) throws Exception {
                //增加的两个解码器
                arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
                arg0.pipeline().addLast(new StringDecoder());
                arg0.pipeline().addLast(new TimeServerHandler());
            }
        }

此时在TimeServerHandler中的channelRead读取到的msg直接强制转换成String，就是删除回车换行符的请求消息，就是我们需要的请求消息。

public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
            String body=(String)msg;
            System.out.println("The time server receive order : "+body
                    +" ; the counter is : " + ++counter);
            String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
            currentTime=currentTime + System.getProperty("line.separator");
            ByteBuf resp =Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.writeAndFlush(resp);
        }

客户端也类似在TimeClient的ChannelInitializer< SocketChannel >也新增LineBasedFrameDecoder和StringDecoder解码器。

public void connect(int port,String host) throws Exception{
            EventLoopGroup group=new NioEventLoopGroup();
            try{
                Bootstrap b=new Bootstrap();
                //Channel需要设置为NioSocketChannel，然后为其添加Handler
                b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>(){
                    //为了简单直接创建匿名内部类，实现initChannel方法
                    //其作用是当创建NioSocketChannel成功之后，在进行初始化时，
                    //将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件
                    @Override
                    public 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();
                //当客户端连接关闭之后，客户端主函数退出，退出前释放NIO线程组的资源
                f.channel().closeFuture().sync();
            }finally{
            }
        }

TimeClientHandler中的channelRead方法读取到的msg也是解码后的应答消息，相比之前的简洁了许多：

public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{
            String body=(String)msg;
            System.out.println("Now is : " + body + " ; the counter is : " + ++counter);
        }

完整源码：  
**TimerClient**：

import io.netty.bootstrap.Bootstrap;
    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;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    public class TimeClient { 
        public void connect(int port,String host) throws Exception{ 
            EventLoopGroup group=new NioEventLoopGroup();
            try{ 
                Bootstrap b=new Bootstrap();
                //Channel需要设置为NioSocketChannel，然后为其添加Handler
                b.group(group).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new ChannelInitializer<SocketChannel>(){ 
                    //为了简单直接创建匿名内部类，实现initChannel方法
                    //其作用是当创建NioSocketChannel成功之后，在进行初始化时，
                    //将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件
                    @Override
                    public 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();
                //当客户端连接关闭之后，客户端主函数退出，退出前释放NIO线程组的资源
                f.channel().closeFuture().sync();
            }finally{ 
    
            }
        }
        public static void main(String[] args) throws Exception { 
            int port=30000;
            new TimeClient().connect(port, "127.0.0.1");
        }
    }

**TimeClientHandler**：

import java.util.logging.Logger;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class TimeClientHandler extends ChannelHandlerAdapter{ 
        private static final Logger logger=Logger.getLogger(TimeClientHandler.class.getName());
        private int counter;
        private byte[] req;
    
        public TimeClientHandler(){ 
            req=("QUERY TIME ORDER" + System.getProperty("line.separator")).getBytes();
        }
        /** * 当客户端和服务器TCP链路建立成功后，NIO线程会调用channelActive方法 */
        @Override
        public void channelActive(ChannelHandlerContext ctx){ 
            //发送查询时间的指令给服务端
            ByteBuf message=null;
            //循环发送100条消息，每发送一条就刷新一次
            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){ 
            logger.warning("Unexpected exception from downstrea : " + cause.getMessage());
            ctx.close();
        }
    }

**TimeServer**：

**TimeServerHandler**：

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class TimeServerHandler extends ChannelHandlerAdapter{ 
        private int counter;
        @Override
        public void channelRead(ChannelHandlerContext ctx,Object msg)throws Exception{ 
            String body=(String)msg;
            System.out.println("The time server receive order : "+body
                    +" ; the counter is : " + ++counter);
            String currentTime="QUERY TIME ORDER".equalsIgnoreCase(body)?new java.util.Date(System.currentTimeMillis()).toString():"BAD ORDER";
            currentTime=currentTime + System.getProperty("line.separator");
    
            ByteBuf resp =Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.writeAndFlush(resp);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx,Throwable cause){ 
            ctx.close();
        }
    }

运行结果：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 1]  
服务端也不会出现粘包现象：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 2]

## 采用分隔符和定长解码器解决TCP粘包和拆包 ##

TCP以流的形式进行数据传输，上层的应用协议为了对消息进行区分，往往采用如下的4种方式。  
（1）消息长度固定，累计读到长度总和为定长len的报文后，就认为读取到了一个完整的消息；然后重新开始读取下一个“完整”的数据包；  
（2）将回车换行符作为消息结束符，如ftp协议；  
（3）将特殊的分隔符作为消息的结束标识，回车换行符就是一种特殊的结束分割符；  
（4）通过在消息头中定义的长度字段表示消息的总长度；  
Netty对以上4种应用做了抽象,提供了4种解码器，有了解码器,码农们不用考虑TCP的粘包、拆包的问题了。

**LineBasedFrameDecoder**：依次编译bytebuf中的可读字符，判断看是否有“\\n”或者“\\r\\n”，如果有，就以此位置为结束位置，从可读索引到结束位置区间的字节就组成了一行。它是以换行符为结束标志的解码器，支持携带结束符或者不携带结束符两种解码方式，同时支持单行的最大长度。如果连续读取到最大长度后，仍然没有发现换行符，就会抛出异常，同时忽略掉之前读到的异常码流。  
**FixedLengthFrameDecoder**:是固定长度解码器，它能按照指定的长度对消息进行自动解码，开发者不需要考虑TCP的粘包等问题。利用FixedLengthFrameDecoder解码，无论一次性接收到多少的数据，他都会按照构造函数中设置的长度进行解码；如果是半包消息，FixedLengthFrameDecoder会缓存半包消息并等待下一个包，到达后进行拼包，直到读取完整的包。

**DelimiterBasedFrameDecoder**：是自定义的分隔符解码，构造函数的第一个参数表示单个消息的最大长度，当达到该长度后仍然没有查到分隔符，就抛出TooLongFrameException异常，防止由于异常码流缺失分隔符导致的内存溢出。  
  **下例中，DelimiterBasedFrameDecoder以$\_作为分隔符，EchoServer.java关键代码**：

@Override
    public void initChannel(SocketChannel ch) throws Exception{
        //创建分隔符缓冲对象ByteBuf，以$_为分隔符
       ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
       //1024表示单条消息的最大长度，当达到该长度后仍然没有查找到分隔符
       //就抛出TooLongFrameException异常
       //第二个参数是分隔符缓冲对象
       new DelimiterBasedFrameDecoder(1024,delimiter));  //后续的ChannelHandler接收到的msg对象将会是完整的消息包
       ch.pipeline().addLast(new StringDecoder()); //将ByteBuf解码成字符串对象 
       ch.pipeline().addLast(new EchoServerHandler());  //接收到的msg消息就是解码后的字符串对象
    }

**EchoServerHandler.java 关键代码**：

@Override
    public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
       String body=(String)msg;
       System.out.println("This is " + ++counter + " times receive client : [" + body + "]");
       body+="$_"; //$_已被过滤掉了，所以这里要拼接上
       ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
       ctx.writeAndFlush(echo);
    }

**EchoClient.java 关键代码**：

@Override 
    public void initChannel(SocketChannel ch) throws Exception{
       ByteBuf delimiter=Unpooled.copiedBuffer("$_".getBytes());
       ch.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter));
       ch.pipeline().addLast(new StringDecoder());
       ch.pipeline().addLast(new EchoClientHandler());
    }

**EchoClientHandler.java 关键代码**：

@Override
    public void channelActive(ChannelHandlerContext ctx){
       for(int i=0;i<10;i++){
          ctx.writeAndFlush(Unpooled.copiedBuffer(ECHOREQ.getBytes()));
       }
    }

**FixedLengthFrameDecoder应用开发**：固定长度解码器，按照指定的长度对消息进行自动解码，开发者不需要考虑TCP的粘包/拆包问题。  
**EchoServer.java 关键代码**：

@Override
    public void initChannel (SocketChannel ch) throws Exception{
       ch.pipeline().addLast(new FixedLengthFrameDecoder(20));
       ch.pipeline().addLast(new StringDecoder());
       ch.pipeline().addLast(new EchoServerHandler()));
    }

**EchoServerHandler.java 关键代码**：

@Override
    public void channelRead(ChannelHandlerContext ctx,Object msg) throws Exception{
       System.out.println("Receive client : [" + msg + "]");
    }

完整代码客户端：  
**EchoClient**：

import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    public class EchoClient { 
        public void connection(int port,String host) throws InterruptedException { 
            NioEventLoopGroup workGroup = new NioEventLoopGroup();
            try { 
                Bootstrap b = new Bootstrap();
                b.group(workGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY,true)
                        .handler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                ByteBuf delimiter = Unpooled.copiedBuffer("#".getBytes());
                                socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter))
                                        .addLast(new StringDecoder())
                                        .addLast(new EchoClientHandler());
    //
                            }
                        });
    // 发起异步连接操作
                ChannelFuture f = b.connect(host,port).sync();
    // 等待客户端链路关闭
                f.channel().closeFuture().sync();
            } finally { 
                workGroup.shutdownGracefully();
            }
        }
        public static void main(String[] args) throws InterruptedException { 
            int port = 30000;
            if(args.length>0&&args!=null){ 
                System.out.println(args[0]);
                port = Integer.parseInt(args[0]);
            }
            new EchoClient().connection(port,"127.0.0.1");
        }
    }

**EchoClientHandler**：

import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    public class EchoClientHandler extends ChannelInboundHandlerAdapter { 
        private int count;
        static final String ECHO_REQ = "hello,MrRight,welcome here!#";
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception { 
            for(int i=0;i<10;i++){ 
                ctx.writeAndFlush(Unpooled.copiedBuffer(ECHO_REQ.getBytes()));
            }
        }
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            String body = (String) msg;
            System.out.println("this is client receive msg"+ ++count +"times:【"+body+"】");
        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            super.channelReadComplete(ctx);
        }
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            super.exceptionCaught(ctx, cause);
        }
    }

客户端：  
**EchoServer**：

import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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.NioServerSocketChannel;
    import io.netty.handler.codec.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    public class EchoServer { 
        public void bind(int port) throws InterruptedException { 
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workGroup = new NioEventLoopGroup();
            try { 
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup,workGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG,100)
                        .childHandler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() { 
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception { 
                                ByteBuf delimiter = Unpooled.copiedBuffer("#".getBytes());//创建一个分隔符，确定为结束标志
                                socketChannel.pipeline().addLast(new DelimiterBasedFrameDecoder(1024,delimiter))
                                        .addLast(new StringDecoder())
                                        .addLast(new EchoServerHandler());
                            }
                        });
    // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally { 
                bossGroup.shutdownGracefully();
                workGroup.shutdownGracefully();
            }
        }
        public static void main(String[] args) throws InterruptedException { 
            int port = 30000;
            if(args.length>0&&args!=null){ 
                port = Integer.parseInt(args[0]);
            }
            new EchoServer().bind(port);
    
        }
    }

**EchoServerHandler**：

import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    public class EchoServerHandler extends ChannelInboundHandlerAdapter { 
        int count;
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
            String body = (String) msg;
            System.out.println("This is"+ ++count +" times server receive client request.");
            body += "#";
            ByteBuf echo = Unpooled.copiedBuffer(body.getBytes());
            ctx.writeAndFlush(echo);
        }
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { 
            ctx.flush();
        }
        @Override![在这里插入图片描述](https://img-blog.csdnimg.cn/20190402191322394.png)
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
            ctx.close();
        }
    }

运行结果：  
![在这里插入图片描述][2019040219131225.png]  
![在这里插入图片描述][20190402191334913.png]

[https_netty.io_downloads.html]: https://netty.io/downloads.html
[20190402104918490.png]: https://img-blog.csdnimg.cn/20190402104918490.png
[20190402104926712.png]: /images/20220225/ff67ac6558bf45aa90b46d2a2ea1787a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70]: /images/20220225/9b803f1b57e149d684523556742c908b.png
[20190402155847879.png]: /images/20220225/e8f0e3b1ff5b4deaaa6c5b36efa738f0.png
[20190402155857352.png]: /images/20220225/a420ec0140754966b2f51b326001d2b3.png
[20190402160504537.png]: /images/20220225/a6407a5de841488eb08f3d0a08a6c079.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 1]: /images/20220225/39879a1f08cd4d9db3779ef1b0f2cd2c.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MTI2MjQ1Mw_size_16_color_FFFFFF_t_70 2]: /images/20220225/effe91f5aec9426682178f9c3d901f58.png
[2019040219131225.png]: /images/20220225/27027b683c264e428da1e460f35f7ae2.png
[20190402191334913.png]: /images/20220225/7a9843d65ac2482682172d16695c74c0.png