BIO、NIO、AIO系列二：Netty

一、概述

Netty是一个Java的开源框架。提供异步的、事件驱动的网络应用程序框架和工具，用以快速开发高性能、高可靠性的网络服务器和客户端程序。

Netty是一个NIO客户端，服务端框架。允许快速简单的开发网络应用程序。例如：服务端和客户端之间的协议，它简化了网络编程规范。

二、NIO开发的问题

1、NIO类库和API复杂，使用麻烦。

2、需要具备Java多线程编程能力（涉及到Reactor模式）。

3、客户端断线重连、网络不稳定、半包读写、失败缓存、网络阻塞和异常码流等问题处理难度非常大

4、存在部分BUG

NIO进行服务器开发的步骤很复杂有以下步骤：

1、创建ServerSocketChannel，配置为非阻塞模式；

2、绑定监听，配置TCP参数；

3、创建一个独立的IO线程，用于轮询多路复用器Selector；

4、创建Selector，将之前创建的ServerSocketChannel注册到Selector上，监听Accept事件；

5、启动IO线程，在循环中执行Select.select()方法，轮询就绪的Channel；

6、当轮询到处于就绪状态的Channel时，需要对其进行判断，如果是OP_ACCEPT状态，说明有新的客户端接入，则调用ServerSocketChannel.accept()方法接受新的客户端；

7、设置新接入的客户端链路SocketChannel为非阻塞模式，配置TCP参数；

8、将SocketChannel注册到Selector上，监听READ事件；

9、如果轮询的Channel为OP_READ，则说明SocketChannel中有新的准备就绪的数据包需要读取，则构造ByteBuffer对象，读取数据包；

10、如果轮询的Channel为OP_WRITE，则说明还有数据没有发送完成，需要继续发送。

三、Netty的优点

1、API使用简单，开发门槛低；

2、功能强大，预置了多种编解码功能，支持多种主流协议；

3、定制功能强，可以通过ChannelHandler对通信框架进行灵活的扩展；

4、性能高，通过与其他业界主流的NIO框架对比，Netty综合性能最优；

5、成熟、稳定，Netty修复了已经发现的NIO所有BUG；

6、社区活跃；

7、经历了很多商用项目的考验。

四、Netty使用demo示例

服务端TimeServer.java

1 1 package com.studyio.netty;
2 2 
3 3 import io.netty.bootstrap.ServerBootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioServerSocketChannel;
1111 
1212 /**
1313  * 
1414  * @author lgs
1515  * 服务端
1616  */
1717 public class TimeServer {
1818 
1919     public static void main(String[] args) throws Exception {
2020         int port=8080; //服务端默认端口
2121         new TimeServer().bind(port);
2222     }
2323     
2424     public void bind(int port) throws Exception{
2525         //1用于服务端接受客户端的连接 线程池里面只有一个线程
2626         EventLoopGroup acceptorGroup = new NioEventLoopGroup(1);
2727         //2用于进行SocketChannel的网络读写 线程池有多个线程
2828         EventLoopGroup workerGroup = new NioEventLoopGroup();
2929         try {
3030             //Netty用于启动NIO服务器的辅助启动类
3131             ServerBootstrap sb = new ServerBootstrap();
3232             //将两个NIO线程组传入辅助启动类中
3333             sb.group(acceptorGroup, workerGroup)
3434                 //设置创建的Channel为NioServerSocketChannel类型
3535                 .channel(NioServerSocketChannel.class)
3636                 //配置NioServerSocketChannel的TCP参数
3737                 .option(ChannelOption.SO_BACKLOG, 1024)
3838                 //设置绑定IO事件的处理类
3939                 .childHandler(new ChannelInitializer<SocketChannel>() {
4040                     //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
4141                     @Override
4242                     protected void initChannel(SocketChannel arg0) throws Exception {
4343                         
4444                         arg0.pipeline().addLast(new TimeServerHandler());
4545                     }
4646                 });
4747             //绑定端口，同步等待成功（sync()：同步阻塞方法，等待bind操作完成才继续）
4848             //ChannelFuture主要用于异步操作的通知回调
4949             ChannelFuture cf = sb.bind(port).sync();
5050             System.out.println("服务端启动在8080端口。");
5151             //等待服务端监听端口关闭
5252             cf.channel().closeFuture().sync();
5353         } finally {
5454             //优雅退出，释放线程池资源
5555             acceptorGroup.shutdownGracefully();
5656             workerGroup.shutdownGracefully();
5757         }
5858     }
5959 }

服务端用于对网络资源进行读写操作TimeServerHandler.java

1 1 package com.studyio.netty;
2 2 
3 3 import java.util.Date;
4 4 
5 5 import io.netty.buffer.ByteBuf;
6 6 import io.netty.buffer.Unpooled;
7 7 import io.netty.channel.ChannelHandlerAdapter;
8 8 import io.netty.channel.ChannelHandlerContext;
9 9 
1010 /**
1111  * 
1212  * @author lgs
1313  * @readme 服务端用于对网络资源进行读写操作，通常我们只需要关注channelRead和exceptionCaught方法。
1414  * 
1515  */
1616 public class TimeServerHandler extends ChannelHandlerAdapter {
1717 
1818     @Override
1919     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
2020         ByteBuf buf = (ByteBuf) msg;
2121         //buf.readableBytes():获取缓冲区中可读的字节数；
2222         //根据可读字节数创建数组
2323         byte[] req = new byte[buf.readableBytes()];
2424         buf.readBytes(req);
2525         String body = new String(req, "UTF-8");
2626         System.out.println("The time server(Thread:"+Thread.currentThread()+") receive order : "+body);
2727         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
2828         
2929         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
3030         //将待发送的消息放到发送缓存数组中
3131         ctx.writeAndFlush(resp);
3232     }
3333 }

客户端TimeClient.java

1 1 package com.studyio.netty;
2 2 
3 3 import io.netty.bootstrap.Bootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioSocketChannel;
1111 
1212 /**
1313  * 
1414  * @author lgs
1515  * 客户端
1616  */
1717 public class TimeClient {
1818     public static void main(String[] args) throws Exception {
1919         int port=8080; //服务端默认端口
2020         new TimeClient().connect(port, "127.0.0.1");
2121     }
2222     public void connect(int port, String host) throws Exception{
2323         //配置客户端NIO线程组
2424         EventLoopGroup group = new NioEventLoopGroup();
2525         try {
2626             Bootstrap bs = new Bootstrap();
2727             bs.group(group)
2828                 .channel(NioSocketChannel.class)
2929                 .option(ChannelOption.TCP_NODELAY, true)
3030                 .handler(new ChannelInitializer<SocketChannel>() {
3131                     @Override
3232                     //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
3333                     protected void initChannel(SocketChannel arg0) throws Exception {
3434                         arg0.pipeline().addLast(new TimeClientHandler());
3535                     }
3636                 });
3737             //发起异步连接操作
3838             ChannelFuture cf = bs.connect(host, port).sync();
3939             //等待客户端链路关闭
4040             cf.channel().closeFuture().sync();
4141         } finally {
4242             //优雅退出，释放NIO线程组
4343             group.shutdownGracefully();
4444         }
4545     }
4646 }

客户端向服务端发送数据和接收服务端数据TimeClientHandler.java

1 1 package com.studyio.netty;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.Unpooled;
5 5 import io.netty.channel.ChannelHandlerAdapter;
6 6 import io.netty.channel.ChannelHandlerContext;
7 7 
8 8 /**
9 9  * 
1010  * @author lgs
1111  * 客户端向服务端发送数据和接收服务端数据
1212  */
1313 public class TimeClientHandler extends ChannelHandlerAdapter {
1414 
1515     @Override
1616     //向服务器发送指令
1717     public void channelActive(ChannelHandlerContext ctx) throws Exception {
1818         for (int i = 0; i < 5; i++) {
1919             byte[] req = "QUERY TIME ORDER".getBytes();
2020             ByteBuf firstMessage = Unpooled.buffer(req.length);
2121             firstMessage.writeBytes(req);
2222             ctx.writeAndFlush(firstMessage);
2323         }
2424     }
2525 
2626     @Override
2727     //接收服务器的响应
2828     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
2929         ByteBuf buf = (ByteBuf) msg;
3030         //buf.readableBytes():获取缓冲区中可读的字节数；
3131         //根据可读字节数创建数组
3232         byte[] req = new byte[buf.readableBytes()];
3333         buf.readBytes(req);
3434         String body = new String(req, "UTF-8");
3535         System.out.println("Now is : "+body);
3636     }
3737 
3838     @Override
3939     //异常处理
4040     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
4141         //释放资源
4242         ctx.close();
4343     }
4444     
4545 }

五、粘包/拆包问题

TCP粘包拆包问题示例图：

说明：

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

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

2、服务端一次接收到了两个数据包，D1和D2粘合在一起，被称为TCP粘包；

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

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

如果此时服务器TCP接收滑窗非常小，而数据包D1和D2比较大，很有可能发生第五种情况，既服务端分多次才能将D1和D2包接收完全，期间发生多次拆包；

总结：

粘包：客户端发送的数据D2,D1可能被合并成一个D2D1发送到服务端

拆包：客户端发送的数据D2,D1可能被拆分成D2_1,D2_2D1或者D2D1_1,D1_2发送到服务端

粘包拆包问题的解决策略

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

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

FixedLengthFrameDecoder

是固定长度解码器，能够按照指定的长度对消息进行自动解码，开发者不需要考虑TCP的粘包/拆包问题。

服务端：

TimeServer.java

1 1 package com.studyio.nettyFixedLength;
2 2 
3 3 import io.netty.bootstrap.ServerBootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioServerSocketChannel;
1111 import io.netty.handler.codec.FixedLengthFrameDecoder;
1212 import io.netty.handler.codec.string.StringDecoder;
1313 
1414 /**
1515  * 
1616  * @author lgs
1717  * 处理粘包/拆包问题-消息定长,固定长度处理
1818  */
1919 public class TimeServer {
2020 
2121     public static void main(String[] args) throws Exception {
2222         int port=8080; //服务端默认端口
2323         new TimeServer().bind(port);
2424     }
2525     public void bind(int port) throws Exception{
2626         //Reactor线程组
2727         //1用于服务端接受客户端的连接
2828         EventLoopGroup acceptorGroup = new NioEventLoopGroup();
2929         //2用于进行SocketChannel的网络读写
3030         EventLoopGroup workerGroup = new NioEventLoopGroup();
3131         try {
3232             //Netty用于启动NIO服务器的辅助启动类
3333             ServerBootstrap sb = new ServerBootstrap();
3434             //将两个NIO线程组传入辅助启动类中
3535             sb.group(acceptorGroup, workerGroup)
3636                 //设置创建的Channel为NioServerSocketChannel类型
3737                 .channel(NioServerSocketChannel.class)
3838                 //配置NioServerSocketChannel的TCP参数
3939                 .option(ChannelOption.SO_BACKLOG, 1024)
4040                 //设置绑定IO事件的处理类
4141                 .childHandler(new ChannelInitializer<SocketChannel>() {
4242                     @Override
4343                     protected void initChannel(SocketChannel arg0) throws Exception {
4444                         //处理粘包/拆包问题-固定长度处理
4545                         arg0.pipeline().addLast(new FixedLengthFrameDecoder(16));
4646                         arg0.pipeline().addLast(new StringDecoder());
4747                         arg0.pipeline().addLast(new TimeServerHandler());
4848                     }
4949                 });
5050             //绑定端口，同步等待成功（sync()：同步阻塞方法）
5151             //ChannelFuture主要用于异步操作的通知回调
5252             ChannelFuture cf = sb.bind(port).sync();
5353                 
5454             //等待服务端监听端口关闭
5555             cf.channel().closeFuture().sync();
5656         } finally {
5757             //优雅退出，释放线程池资源
5858             acceptorGroup.shutdownGracefully();
5959             workerGroup.shutdownGracefully();
6060         }
6161     }
6262 }

TimeServerHandler.java

1 1 package com.studyio.nettyFixedLength;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.Unpooled;
5 5 import io.netty.channel.ChannelHandlerAdapter;
6 6 import io.netty.channel.ChannelHandlerContext;
7 7 
8 8 /**
9 9  * 
1010  * @author lgs
1111  * @readme 用于对网络时间进行读写操作，通常我们只需要关注channelRead和exceptionCaught方法。
1212  * 
1313  */
1414 public class TimeServerHandler extends ChannelHandlerAdapter {
1515 
1616     private int counter;
1717     @Override
1818     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
1919         String body = (String) msg;
2020         System.out.println("The time server(Thread:"+Thread.currentThread()+") receive order : "+body+". the counter is : "+ ++counter);
2121         String currentTime = body;
2222         
2323         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
2424         
2525         //将待发送的消息放到发送缓存数组中
2626         ctx.writeAndFlush(resp);
2727     }
2828 }

客户端：

TimeClient.java

1 1 package com.studyio.nettyFixedLength;
2 2 
3 3 import io.netty.bootstrap.Bootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioSocketChannel;
1111 import io.netty.handler.codec.FixedLengthFrameDecoder;
1212 import io.netty.handler.codec.string.StringDecoder;
1313 
1414 /**
1515  * 
1616  * @author lgs
1717  * 处理粘包/拆包问题-消息定长,固定长度处理
1818  */
1919 public class TimeClient {
2020     public static void main(String[] args) throws Exception {
2121         int port=8080; //服务端默认端口
2222         new TimeClient().connect(port, "127.0.0.1");
2323     }
2424     public void connect(int port, String host) throws Exception{
2525         //配置客户端NIO线程组
2626         EventLoopGroup group = new NioEventLoopGroup();
2727         try {
2828             Bootstrap bs = new Bootstrap();
2929             bs.group(group)
3030                 .channel(NioSocketChannel.class)
3131                 .option(ChannelOption.TCP_NODELAY, true)
3232                 .handler(new ChannelInitializer<SocketChannel>() {
3333                     @Override
3434                     //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
3535                     protected void initChannel(SocketChannel arg0) throws Exception {
3636                         //处理粘包/拆包问题-固定长度处理
3737                         arg0.pipeline().addLast(new FixedLengthFrameDecoder(16));
3838                         arg0.pipeline().addLast(new StringDecoder());
3939                         
4040                         arg0.pipeline().addLast(new TimeClientHandler());
4141                     }
4242                 });
4343             //发起异步连接操作
4444             ChannelFuture cf = bs.connect(host, port).sync();
4545             //等待客户端链路关闭
4646             cf.channel().closeFuture().sync();
4747         } finally {
4848             //优雅退出，释放NIO线程组
4949             group.shutdownGracefully();
5050         }
5151     }
5252 }

TimeClientHandler.java

1 1 package com.studyio.nettyFixedLength;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.Unpooled;
5 5 import io.netty.channel.ChannelHandlerAdapter;
6 6 import io.netty.channel.ChannelHandlerContext;
7 7 
8 8 /**
9 9  * 
1010  * @author lgs
1111  * 
1212  */
1313 public class TimeClientHandler extends ChannelHandlerAdapter {
1414 
1515     private int counter;
1616     private byte[] req;
1717     
1818     @Override
1919     //向服务器发送指令
2020     public void channelActive(ChannelHandlerContext ctx) throws Exception {
2121         ByteBuf message=null;
2222         //模拟一百次请求，发送重复内容
2323         for (int i = 0; i < 100; i++) {
2424             req = ("QUERY TIME ORDER").getBytes();
2525             message=Unpooled.buffer(req.length);
2626             message.writeBytes(req);
2727             ctx.writeAndFlush(message);
2828         }
2929     }
3030 
3131     @Override
3232     //接收服务器的响应
3333     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3434         String body = (String) msg;
3535         System.out.println("Now is : "+body+". the counter is : "+ ++counter);
3636     }
3737 
3838     @Override
3939     //异常处理
4040     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
4141         //释放资源
4242         ctx.close();
4343     }
4444     
4545 }

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

2.1 处理粘包/拆包问题-回车换行符进行分割

LineBasedFrameDecoder

服务端：

TimeServer.java

1 1 package com.studyio.nettyLine;
2 2 
3 3 import io.netty.bootstrap.ServerBootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioServerSocketChannel;
1111 import io.netty.handler.codec.LineBasedFrameDecoder;
1212 import io.netty.handler.codec.string.StringDecoder;
1313 
1414 /**
1515  * 
1616  * @author lgs
1717  * 处理粘包/拆包问题-回车换行符进行分割
1818  * 
1919  */
2020 public class TimeServer {
2121 
2222     public static void main(String[] args) throws Exception {
2323         int port=8080; //服务端默认端口
2424         new TimeServer().bind(port);
2525     }
2626     public void bind(int port) throws Exception{
2727         //Reactor线程组
2828         //1用于服务端接受客户端的连接
2929         EventLoopGroup acceptorGroup = new NioEventLoopGroup();
3030         //2用于进行SocketChannel的网络读写
3131         EventLoopGroup workerGroup = new NioEventLoopGroup();
3232         try {
3333             //Netty用于启动NIO服务器的辅助启动类
3434             ServerBootstrap sb = new ServerBootstrap();
3535             //将两个NIO线程组传入辅助启动类中
3636             sb.group(acceptorGroup, workerGroup)
3737                 //设置创建的Channel为NioServerSocketChannel类型
3838                 .channel(NioServerSocketChannel.class)
3939                 //配置NioServerSocketChannel的TCP参数
4040                 .option(ChannelOption.SO_BACKLOG, 1024)
4141                 //设置绑定IO事件的处理类
4242                 .childHandler(new ChannelInitializer<SocketChannel>() {
4343                     @Override
4444                     protected void initChannel(SocketChannel arg0) throws Exception {
4545                         //处理粘包/拆包问题-换行符处理
4646                         arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
4747                         arg0.pipeline().addLast(new StringDecoder());
4848                         
4949                         arg0.pipeline().addLast(new TimeServerHandler());
5050                     }
5151                 });
5252             //绑定端口，同步等待成功（sync()：同步阻塞方法）
5353             //ChannelFuture主要用于异步操作的通知回调
5454             ChannelFuture cf = sb.bind(port).sync();
5555                 
5656             //等待服务端监听端口关闭
5757             cf.channel().closeFuture().sync();
5858         } finally {
5959             //优雅退出，释放线程池资源
6060             acceptorGroup.shutdownGracefully();
6161             workerGroup.shutdownGracefully();
6262         }
6363     }
6464 }

TimeServerHandler.java

1 1 package com.studyio.nettyLine;
2 2 
3 3 import java.util.Date;
4 4 
5 5 import io.netty.buffer.ByteBuf;
6 6 import io.netty.buffer.Unpooled;
7 7 import io.netty.channel.ChannelHandlerAdapter;
8 8 import io.netty.channel.ChannelHandlerContext;
9 9 
1010 /**
1111  * 
1212  * @author lgs
1313  * @readme 用于对网络时间进行读写操作，通常我们只需要关注channelRead和exceptionCaught方法。
1414  * 处理粘包/拆包问题-回车换行符进行分割
1515  */
1616 public class TimeServerHandler extends ChannelHandlerAdapter {
1717 
1818     private int counter;
1919     @Override
2020     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
2121 //        ByteBuf buf = (ByteBuf) msg;
2222 //        //buf.readableBytes():获取缓冲区中可读的字节数；
2323 //        //根据可读字节数创建数组
2424 //        byte[] req = new byte[buf.readableBytes()];
2525 //        buf.readBytes(req);
2626 //        String body = new String(req, "UTF-8");
2727         String body = (String) msg;
2828         System.out.println("The time server(Thread:"+Thread.currentThread()+") receive order : "+body+". the counter is : "+ ++counter);
2929         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
3030         currentTime = currentTime + System.getProperty("line.separator");
3131         
3232         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
3333         //将待发送的消息放到发送缓存数组中
3434         ctx.writeAndFlush(resp);
3535     }
3636 
3737 }

客户端：

TimeClient.java

1 1 package com.studyio.nettyLine;
2 2 
3 3 import io.netty.bootstrap.Bootstrap;
4 4 import io.netty.channel.ChannelFuture;
5 5 import io.netty.channel.ChannelInitializer;
6 6 import io.netty.channel.ChannelOption;
7 7 import io.netty.channel.EventLoopGroup;
8 8 import io.netty.channel.nio.NioEventLoopGroup;
9 9 import io.netty.channel.socket.SocketChannel;
1010 import io.netty.channel.socket.nio.NioSocketChannel;
1111 import io.netty.handler.codec.LineBasedFrameDecoder;
1212 import io.netty.handler.codec.string.StringDecoder;
1313 
1414 /**
1515  * 
1616  * @author lgs
1717  * 处理粘包/拆包问题-回车换行符进行分割
1818  * 
1919  */
2020 public class TimeClient {
2121     public static void main(String[] args) throws Exception {
2222         int port=8080; //服务端默认端口
2323         new TimeClient().connect(port, "127.0.0.1");
2424     }
2525     public void connect(int port, String host) throws Exception{
2626         //配置客户端NIO线程组
2727         EventLoopGroup group = new NioEventLoopGroup();
2828         try {
2929             Bootstrap bs = new Bootstrap();
3030             bs.group(group)
3131                 .channel(NioSocketChannel.class)
3232                 .option(ChannelOption.TCP_NODELAY, true)
3333                 .handler(new ChannelInitializer<SocketChannel>() {
3434                     @Override
3535                     //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
3636                     protected void initChannel(SocketChannel arg0) throws Exception {
3737                         //处理粘包/拆包问题-换行符处理
3838                         arg0.pipeline().addLast(new LineBasedFrameDecoder(1024));
3939                         arg0.pipeline().addLast(new StringDecoder());
4040                         
4141                         arg0.pipeline().addLast(new TimeClientHandler());
4242                     }
4343                 });
4444             //发起异步连接操作
4545             ChannelFuture cf = bs.connect(host, port).sync();
4646             //等待客户端链路关闭
4747             cf.channel().closeFuture().sync();
4848         } finally {
4949             //优雅退出，释放NIO线程组
5050             group.shutdownGracefully();
5151         }
5252     }
5353 }

TimeClientHandler.java

1 1 package com.studyio.nettyLine;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.Unpooled;
5 5 import io.netty.channel.ChannelHandlerAdapter;
6 6 import io.netty.channel.ChannelHandlerContext;
7 7 
8 8 /**
9 9  * 
1010  * @author lgs
1111  * 处理粘包/拆包问题-回车换行符进行分割
1212  */
1313 public class TimeClientHandler extends ChannelHandlerAdapter {
1414 
1515     private int counter;
1616     private byte[] req;
1717     
1818     @Override
1919     //向服务器发送指令
2020     public void channelActive(ChannelHandlerContext ctx) throws Exception {
2121         ByteBuf message=null;
2222         //模拟一百次请求，发送重复内容
2323         for (int i = 0; i < 200; i++) {
2424             //回车换行符System.getProperty("line.separator")
2525             req = ("QUERY TIME ORDER"+System.getProperty("line.separator")).getBytes();
2626             message=Unpooled.buffer(req.length);
2727             message.writeBytes(req);
2828             ctx.writeAndFlush(message);
2929         }
3030         
3131     }
3232 
3333     @Override
3434     //接收服务器的响应
3535     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3636 //        ByteBuf buf = (ByteBuf) msg;
3737 //        //buf.readableBytes():获取缓冲区中可读的字节数；
3838 //        //根据可读字节数创建数组
3939 //        byte[] req = new byte[buf.readableBytes()];
4040 //        buf.readBytes(req);
4141 //        String body = new String(req, "UTF-8");
4242         String body = (String) msg;
4343         System.out.println("Now is : "+body+". the counter is : "+ ++counter);
4444     }
4545 
4646     @Override
4747     //异常处理
4848     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
4949         //释放资源
5050         ctx.close();
5151     }
5252     
5353 }

2.2 处理粘包/拆包问题自定义分隔符****进行分割

DelimiterBasedFrameDecoder

实现自定义分隔符作为消息的结束标志，完成解码。

服务端：

TimeServer.java

1 1 package com.studyio.nettyDelimiter;
2 2 
3 3 import io.netty.bootstrap.ServerBootstrap;
4 4 import io.netty.buffer.ByteBuf;
5 5 import io.netty.buffer.Unpooled;
6 6 import io.netty.channel.ChannelFuture;
7 7 import io.netty.channel.ChannelInitializer;
8 8 import io.netty.channel.ChannelOption;
9 9 import io.netty.channel.EventLoopGroup;
1010 import io.netty.channel.nio.NioEventLoopGroup;
1111 import io.netty.channel.socket.SocketChannel;
1212 import io.netty.channel.socket.nio.NioServerSocketChannel;
1313 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
1414 import io.netty.handler.codec.string.StringDecoder;
1515 
1616 /**
1717  * 
1818  * @author lgs
1919  * 处理粘包/拆包问题-定义分隔符
2020  * 
2121  */
2222 public class TimeServer {
2323     public static void main(String[] args) throws Exception {
2424         int port=8080; //服务端默认端口
2525         new TimeServer().bind(port);
2626     }
2727 
2828     public void bind(int port) throws Exception{
2929         //Reactor线程组
3030         //1用于服务端接受客户端的连接
3131         EventLoopGroup acceptorGroup = new NioEventLoopGroup();
3232         //2用于进行SocketChannel的网络读写
3333         EventLoopGroup workerGroup = new NioEventLoopGroup();
3434         try {
3535             //Netty用于启动NIO服务器的辅助启动类
3636             ServerBootstrap sb = new ServerBootstrap();
3737             //将两个NIO线程组传入辅助启动类中
3838             sb.group(acceptorGroup, workerGroup)
3939                 //设置创建的Channel为NioServerSocketChannel类型
4040                 .channel(NioServerSocketChannel.class)
4141                 //配置NioServerSocketChannel的TCP参数
4242                 .option(ChannelOption.SO_BACKLOG, 1024)
4343                 //设置绑定IO事件的处理类
4444                 .childHandler(new ChannelInitializer<SocketChannel>() {
4545                     @Override
4646                     protected void initChannel(SocketChannel arg0) throws Exception {
4747                         //处理粘包/拆包问题-自定义分隔符处理
4848                         ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
4949                         arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
5050                         arg0.pipeline().addLast(new StringDecoder());
5151                         arg0.pipeline().addLast(new TimeServerHandler());
5252                     }
5353                 });
5454             //绑定端口，同步等待成功（sync()：同步阻塞方法）
5555             //ChannelFuture主要用于异步操作的通知回调
5656             ChannelFuture cf = sb.bind(port).sync();
5757                 
5858             //等待服务端监听端口关闭
5959             cf.channel().closeFuture().sync();
6060         } finally {
6161             //优雅退出，释放线程池资源
6262             acceptorGroup.shutdownGracefully();
6363             workerGroup.shutdownGracefully();
6464         }
6565     }
6666 }

TimeServerHandler.java

1 1 package com.studyio.nettyDelimiter;
2 2 
3 3 import java.util.Date;
4 4 
5 5 import io.netty.buffer.ByteBuf;
6 6 import io.netty.buffer.Unpooled;
7 7 import io.netty.channel.ChannelHandlerAdapter;
8 8 import io.netty.channel.ChannelHandlerContext;
9 9 
1010 /**
1111  * 
1212  * @author lgs
1313  * 处理粘包/拆包问题-定义分隔符
1414  * @readme 用于对网络时间进行读写操作，通常我们只需要关注channelRead和exceptionCaught方法。
1515  * 
1616  */
1717 public class TimeServerHandler extends ChannelHandlerAdapter {
1818 
1919     private int counter;
2020     @Override
2121     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
2222         String body = (String) msg;
2323         System.out.println("The time server(Thread:"+Thread.currentThread()+") receive order : "+body+". the counter is : "+ ++counter);
2424         String currentTime = "QUERY TIME ORDER".equalsIgnoreCase(body) ? new Date(System.currentTimeMillis()).toString() : "BAD ORDER";
2525         //处理粘包/拆包问题-定义分隔符
2626         currentTime += "$_";
2727         
2828         ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
2929         //将待发送的消息放到发送缓存数组中
3030         ctx.writeAndFlush(resp);
3131     }
3232 }

客户端：

TimeClient.java

1 1 package com.studyio.nettyDelimiter;
2 2 
3 3 import io.netty.bootstrap.Bootstrap;
4 4 import io.netty.buffer.ByteBuf;
5 5 import io.netty.buffer.Unpooled;
6 6 import io.netty.channel.ChannelFuture;
7 7 import io.netty.channel.ChannelInitializer;
8 8 import io.netty.channel.ChannelOption;
9 9 import io.netty.channel.EventLoopGroup;
1010 import io.netty.channel.nio.NioEventLoopGroup;
1111 import io.netty.channel.socket.SocketChannel;
1212 import io.netty.channel.socket.nio.NioSocketChannel;
1313 import io.netty.handler.codec.DelimiterBasedFrameDecoder;
1414 import io.netty.handler.codec.string.StringDecoder;
1515 
1616 /**
1717  * 
1818  * @author lgs
1919  * 处理粘包/拆包问题-定义分隔符
2020  * 
2121  */
2222 public class TimeClient {
2323     public static void main(String[] args) throws Exception {
2424         int port=8080; //服务端默认端口
2525         new TimeClient().connect(port, "127.0.0.1");
2626     }
2727     public void connect(int port, String host) throws Exception{
2828         //配置客户端NIO线程组
2929         EventLoopGroup group = new NioEventLoopGroup();
3030         try {
3131             Bootstrap bs = new Bootstrap();
3232             bs.group(group)
3333                 .channel(NioSocketChannel.class)
3434                 .option(ChannelOption.TCP_NODELAY, true)
3535                 .handler(new ChannelInitializer<SocketChannel>() {
3636                     @Override
3737                     //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
3838                     protected void initChannel(SocketChannel arg0) throws Exception {
3939                         //处理粘包/拆包问题-自定义分隔符处理
4040                         ByteBuf delimiter = Unpooled.copiedBuffer("$_".getBytes());
4141                         arg0.pipeline().addLast(new DelimiterBasedFrameDecoder(1024, delimiter));
4242                         arg0.pipeline().addLast(new StringDecoder());
4343                         
4444                         arg0.pipeline().addLast(new TimeClientHandler());
4545                     }
4646                 });
4747             //发起异步连接操作
4848             ChannelFuture cf = bs.connect(host, port).sync();
4949             //等待客户端链路关闭
5050             cf.channel().closeFuture().sync();
5151         } finally {
5252             //优雅退出，释放NIO线程组
5353             group.shutdownGracefully();
5454         }
5555     }
5656 }

TimeClientHandler.java

1 1 package com.studyio.nettyDelimiter;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.Unpooled;
5 5 import io.netty.channel.ChannelHandlerAdapter;
6 6 import io.netty.channel.ChannelHandlerContext;
7 7 
8 8 /**
9 9  * 
1010  * @author lgs
1111  * 处理粘包/拆包问题-定义分隔符
1212  * 
1313  */
1414 public class TimeClientHandler extends ChannelHandlerAdapter {
1515     
1616     private int counter;
1717     private byte[] req;
1818     
1919     @Override
2020     //向服务器发送指令
2121     public void channelActive(ChannelHandlerContext ctx) throws Exception {
2222         ByteBuf message=null;
2323         //模拟一百次请求，发送重复内容
2424         for (int i = 0; i < 200; i++) {
2525             //处理粘包/拆包问题-定义分隔符
2626             req = ("QUERY TIME ORDER"+"$_").getBytes();
2727             message=Unpooled.buffer(req.length);
2828             message.writeBytes(req);
2929             ctx.writeAndFlush(message);
3030         }
3131         
3232     }
3333 
3434     @Override
3535     //接收服务器的响应
3636     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
3737         String body = (String) msg;
3838         System.out.println("Now is : "+body+". the counter is : "+ ++counter);
3939     }
4040 
4141     @Override
4242     //异常处理
4343     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
4444         //释放资源
4545         ctx.close();
4646     }
4747     
4848 }

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

4、更复杂的应用层协议；

六、Netty高性能的原因

1、异步非阻塞通信

在IO编程过程中，当需要同时处理多个客户端接入请求时，可以利用多线程或者IO多路复用技术进行处理。IO多路复用技术通过把多个IO的阻塞复用到同一个Selector的阻塞上，从而使得系统在单线程的情况下可以同时处理多个客户端请求。与传统的多线程/多进程模型相比，IO多路复用的最大优势是系统开销小，系统不需要创建新的额外进程或者线程，也不需要维护这些进程和线程的运行，降低了系统的维护工作量，节省了系统资源。

Netty的IO线程NioEventLoop由于聚合了多路复用器Selector，可以同时并发处理成百上千个客户端SocketChannel。由于读写操作都是非阻塞的，这就可以充分提升IO线程的运行效率，避免由频繁的IO阻塞导致的线程挂起。另外，由于Netty采用了异步通信模式，一个IO线程可以并发处理N个客户端连接和读写操作，这从根本上解决了传统同步阻塞IO中一连接一线程模型，架构的性能、弹性伸缩能力和可靠性都得到了极大的提升。

2、高效的Reactor线程模型

常用的Reactor线程模型有三种，分别如下：

Reactor单线程模型：

Reactor单线程模型，指的是所有的IO操作都在同一个NIO线程上面完成，NIO线程职责如下：

1、作为NIO服务端，接收客户端的TCP连接；

2、作为NIO客户端，向服务端发起TCP连接；

3、读取通信对端的请求或者应答消息；

4、向通信对端发送请求消息或者应答消息；

由于Reactor模式使用的是异步非阻塞IO，所有的IO操作都不会导致阻塞，理论上一个线程可以独立处理所有IO相关操作。从架构层面看，一个NIO线程确实可以完成其承担的职责。例如，通过Acceptor接收客户端的TCP连接请求消息，链路建立成功之后，通过Dispatch将对应的ByteBuffer派发到指定的Handler上进行消息编码。用户Handler可以通过NIO线程将消息发送给客户端。

对于一些小容量应用场景，可以使用单线程模型，但是对于高负载、大并发的应用却不合适，主要原因如下：

1、一个NIO线程同时处理成百上千的链路，性能上无法支撑。即便NIO线程的CPU负荷达到100%，也无法满足海量消息的编码、解码、读取和发送；

2、当NIO线程负载过重后，处理速度将变慢，这会导致大量客户端连接超时，超时之后往往会进行重发，这更加重了NIO线程的负载，最终会导致大量消息积压和处理超时，NIO线程会成为系统的性能瓶颈；

3、可靠性问题。一旦NIO线程意外进入死循环，会导致整个系统通信模块不可用，不能接收和处理外部消息，造成节点故障。

为了解决这些问题，从而演进出了Reactor多线程模型。

Reactor多线程模型：

Reactor多线程模型与单线程模型最大的区别就是有一组NIO线程处理IO操作，特点如下：

1、有一个专门的NIO线程——Acceptor线程用于监听服务端，接收客户端TCP连接请求；

2、网络IO操作——读、写等由一个NIO线程池负责，线程池可以采用标准的JDK线程池实现，它包含一个任务队列和N个可用的线程，由这些NIO线程负责消息的读取、编码、解码和发送；

3、1个NIO线程可以同时处理N条链路，但是1个链路只对应1个NIO线程，防止发生并发操作问题。

在绝大多数场景下，Reactor多线程模型都可以满足性能需求；但是，在极特殊应用场景中，一个NIO线程负责监听和处理所有的客户端连接可能会存在性能问题。例如百万客户端并发连接，或者服务端需要对客户端的握手消息进行安全认证，认证本身非常损耗性能。在这类场景下，单独一个Acceptor线程可能会存在性能不足问题，为了解决性能问题，产生了第三种Reactor线程模型——主从Reactor多线程模型。

主从Reactor多线程模型：

主从Reactor线程模型的特点是：服务端用于接收客户端连接的不再是一个单独的NIO线程，而是一个独立的NIO线程池。Acceptor接收到客户端TCP连接请求处理完成后（可能包含接入认证等），将新创建的SocketChannel注册到IO线程池（subReactor线程池）的某个IO线程上，由它负责SocketChannel的读写和编解码工作。Acceptor线程池只用于客户端的登录、握手和安全认证，一旦链路建立成功，就将链路注册到后端subReactor线程池的IO线程上，由IO线程负责后续的IO操作。

利用主从NIO线程模型，可以解决1个服务端监听线程无法有效处理所有客户端连接的性能不足问题。Netty官方推荐使用该线程模型。它的工作流程总结如下：

1、从主线程池中随机选择一个Reactor线程作为Acceptor线程，用于绑定监听端口，接收客户端连接；

2、Acceptor线程接收客户端连接请求之后，创建新的SocketChannel，将其注册到主线程池的其他Reactor线程上，由其负责接入认证、IP黑白名单过滤、握手等操作；

3、然后也业务层的链路正式建立成功，将SocketChannel从主线程池的Reactor线程的多路复用器上摘除，重新注册到Sub线程池的线程上，用于处理IO的读写操作。

3、无锁化的串行设计

在大多数场景下，并行多线程处理可以提升系统的并发性能。但是，如果对于共享资源的并发访问处理不当，会带来严重的锁竞争，这最终会导致性能的下降。为了尽可能地避免锁竞争带来的性能损耗，可以通过串行化设计，既消息的处理尽可能在同一个线程内完成，期间不进行线程切换，这样就避免了多线程竞争和同步锁。

为了尽可能提升性能，Netty采用了串行无锁化设计，在IO线程内部进行串行操作，避免多线程竞争导致的性能下降。表面上看，串行化设计似乎CPU利用率不高，并发程度不够。但是，通过调整NIO线程池的线程参数，可以同时启动多个串行化的线程并行运行，这种局部无锁化的串行线程设计相比一个队列——多个工作线程模型性能更优。

Netty串行化设计工作原理图如下：

Netty的NioEventLoop读取到消息后，直接调用ChannelPipeline的fireChannelRead(Object msg)，只要用户不主动切换线程，一直会由NioEventLoop调用到用户的Handler，期间不进行线程切换。这种串行化处理方式避免了多线程导致的锁竞争，从性能角度看是最优的。

4、高效的并发编程

Netty中高效并发编程主要体现：

1、volatile的大量、正确使用；

2、CAS和原子类的广泛使用；

3、线程安全容器的使用；

4、通过读写锁提升并发性能。

5、高性能的序列化框架

影响序列化性能的关键因素总结如下：

1、序列化后的码流大小（网络宽带的占用）；

2、序列化与反序列化的性能（CPU资源占用）；

3、是否支持跨语言（异构系统的对接和开发语言切换）。

Netty默认提供了对GoogleProtobuf的支持，通过扩展Netty的编解码接口，用户可以实现其他的高性能序列化框架

6、零拷贝

Netty的“零拷贝”主要体现在三个方面：

1）、Netty的接收和发送ByteBuffer采用DIRECT BUFFERS，使用堆外直接内存进行Socket读写，不需要进行字节缓冲区的二次拷贝。如果使用传统的堆内存（HEAP BUFFERS）进行Socket读写，JVM会将堆内存Buffer拷贝一份到直接内存中，然后才写入Socket中。相比于堆外直接内存，消息在发送过程中多了一次缓冲区的内存拷贝。

2）、第二种“零拷贝 ”的实现CompositeByteBuf，它对外将多个ByteBuf封装成一个ByteBuf，对外提供统一封装后的ByteBuf接口。

3）、第三种“零拷贝”就是文件传输，Netty文件传输类DefaultFileRegion通过transferTo方法将文件发送到目标Channel中。很多操作系统直接将文件缓冲区的内容发送到目标Channel中，而不需要通过循环拷贝的方式，这是一种更加高效的传输方式，提升了传输性能，降低了CPU和内存占用，实现了文件传输的“零拷贝”。

7、内存池

随着JVM虚拟机和JIT即时编译技术的发展，对象的分配和回收是个非常轻量级的工作。但是对于缓冲区Buffer，情况却稍有不同，特别是对于堆外直接内存的分配和回收，是一件耗时的操作。为了尽量重用缓冲区，Netty提供了基于内存池的缓冲区重用机制。

1 1 package com.studyio.netty;
2 2 
3 3 import io.netty.buffer.ByteBuf;
4 4 import io.netty.buffer.PooledByteBufAllocator;
5 5 import io.netty.buffer.Unpooled;
6 6 
7 7 /**
8 8  * 
9 9  * @author lgs
1010  * 通过内存池的方式构建直接缓冲区
1111  */
1212 public class PooledByteBufDemo {
1313 
1414     public static void main(String[] args) {
1515         byte[] content = new byte[1024];
1616         int loop = 3000000;
1717         long startTime = System.currentTimeMillis();
1818         
1919         ByteBuf poolBuffer = null;
2020         for (int i = 0; i < loop; i++) {
2121             //通过内存池的方式构建直接缓冲区
2222             poolBuffer = PooledByteBufAllocator.DEFAULT.directBuffer(1024);
2323             poolBuffer.writeBytes(content);
2424             //释放buffer
2525             poolBuffer.release();
2626         }
2727         long startTime2 = System.currentTimeMillis();
2828         ByteBuf buffer = null;
2929         for (int i = 0; i < loop; i++) {
3030             //通过非内存池的方式构建直接缓冲区
3131             buffer = Unpooled.directBuffer(1024);
3232             buffer.writeBytes(content);
3333             buffer.release();
3434         }
3535         long endTime = System.currentTimeMillis();
3636         System.out.println("The PooledByteBuf use time :"+(startTime2-startTime));
3737         System.out.println("The UnpooledByteBuf use time :"+(endTime-startTime2));
3838     }
3939 }

运行结果：内存池的方式构建直接缓冲区效率更高

The PooledByteBuf use time :740
The UnpooledByteBuf use time :1025

8、灵活的TCP参数配置能力

Netty在启动辅助类中可以灵活的配置TCP参数，满足不同的用户场景。合理设置TCP参数在某些场景下对于性能的提升可以起到的显著的效果，总结一下对性能影响比较大的几个配置项：

1）、SO_RCVBUF和SO_SNDBUF：通常建议值为128KB或者256KB;

2）、TCP_NODELAY：NAGLE算法通过将缓冲区内的小封包自动相连，组成较大的封包，阻止大量小封包的发送阻塞网络，从而提高网络应用效率。但是对于时延敏感的应用场景需要关闭该优化算法；

3）、软中断：如果Linux内核版本支持RPS(2.6.35以上版本)，开启RPS后可以实现软中断，提升网络吞吐量。RPS根据数据包的源地址，目的地址以及目的和源端口，计算出一个hash值，然后根据这个hash值来选择软中断运行的CPU。从上层来看，也就是说将每个连接和CPU绑定，并通过这个hash值，来均衡软中断在多个CPU上，提升网络并行处理性能。

BIO、NIO、AIO系列二：Netty

一、概述

二、NIO开发的问题

三、Netty的优点

四、Netty使用demo示例

五、粘包/拆包问题

1、异步非阻塞通信

2、高效的Reactor线程模型

3、无锁化的串行设计

4、高效的并发编程

5、高性能的序列化框架

6、零拷贝

7、内存池

8、灵活的TCP参数配置能力

评论区

相关推荐