Netty高并发网络应用框架-总结
针对传统阻塞I/O服务模型的 2 个缺点，解决方案： 
主要的基于事件监听的方法： 
常用的方法： 
Netty使用自定义协议+编码解码 来解决Tcp粘包和拆包问题。关键就是要解决服务器每次读取数据长度的问题 ，这个解决就不会出现多读或者少读的问题，从而避免了TCP粘包拆包问题。 核心代码： 
使用场景： 阿里的Dubbo、Google的gRPC、Go语言的rpcx、Apache的thrift、Spring Cloud 等等。
发布日期：2021-05-07 23:39:09
浏览次数：31
分类：精选文章
本文共 39078 字，大约阅读时间需要 130 分钟。
Netty高并发网络应用框架
如想了解更多更全面的Java必备内容可以阅读：所有JAVA必备知识点面试题文章目录：
来…直接进入主题：
Netty网易云课堂在线学习资料：https://study.163.com/course/courseMain.htm?courseId=1209596850文章目录
1、列举一下原生NIO还存的哪些问题？
- NIO类库和API繁杂，使用相比麻烦，需要熟练掌握selector、XXXBuffer、XXXchannel等。
- 需要具备其他技能：比如java多线程编程、网络编程、Reactor模式等等。
- 开发工作量难度都非常大，例如：客户端面临断连重连、网络闪断、半包读写、失败缓存、网络拥塞和异常流的处理等等。
- JDK NIO 的Bug，例如：Epoll bug，它会导致selector空轮询，最终导致CPU 占用100% 等。
2、Netty的介绍，什么是Netty？
Netty是由JBOSS提供的一个异步的、基于事件驱动的网络应用开源框架。用于快速开发 可维护的 高性能 协议服务器和客户端。
Netty是底层完全基于NIO实现 的。提高了吞吐量，降低了延迟；减少了资源消耗；减少了不必要的内存复制。 3、目前存在的线程模型有哪些？
- 传统阻塞I/O服务模型
- Reactor模型
4、Reactor 模型对传统阻塞IO服务模型做了那些改良？
传统阻塞IO服务模型，简略版如下：
针对传统阻塞I/O服务模型的 2 个缺点，解决方案： - 基于I/O复用摸型：多个连接共用一个阻塞对象，应用程序只需要在一个阻塞对象等待，无需阻塞等待所有连接。当某个连接有新的数据可以处理时，操作系统通知应用程序，线程从阻塞状态返回，开始进行业务处理。
- 基于线程池复用线程资源：不必再为每个连接创建线程，将连接完成后的业务处理任务分配给线程进行处理，一个线程可以处理多个连接的业务。
Reactor 模式，简略版如下：
说明：Reactor在一个单独的线程中运行，负责监听和分发事件。- Reactor模式，通过一个或多个输入同时传递给服务处理器的模式(基于事件驱动)。
- 服务器端程序处理传入的多个请求，并将它们同步分派到相应的处理线程。
- Reactor 模式使用IO复用监听事件，收到事件后分发给某个线程(进程)，这就是网络服务器高并发处理的关键。
5、Reactor有哪三种典型的实现?
- 单Reactor单线程
- 单Reactor多线程
- 主从Reactor多线程
6、什么是单Reactor单线程？
工作原理和简单示意图：
- select可以实现应用程序通过一个阻塞对象监听多路连接请求。
- Reactor对象通过select监控客户端请求事件，收到事件后通过dispatch进行分发。
- 如果是建立连接请求，则由Acceptor通过Accept处理连接请求。
- 如果不是建立连接请求，会分发给响应的handler进行处理(read/业务处处理/send 等)
7、什么是单Reactor多线程？
工作原理和简单示意图：
- Reactor对象通过select监控客户端请求事件，收到事件后通过dispatch进行分发。
- 如果是建立连接请求，则由Acceptor通过Accept处理连接请求，然后创建一个handler对象处理完成连接口的各种事件。
- 如果不是建立连接请求，会分发给响应的handler进行处理
- handler只负责响应事件，不做具体的业务处理，通过read读取数据后，分发给后面的worker线程池做业务处理。
- worker线程池会分配给具体的线程完成真正的业务处理，并将结果返回给handler。
- handler收到响应后，通过send将处理结果返回给client。
8、什么是主从Reactor多线程？
工作原理和简单示意图：
注：Reactor主线程可以对应多个Reactor子线程，即MainReactor可以对应多个SubReactor。- Reactor主线程MainReactor对象通过select监听建立连接事件，收到事件后，通过Acceptor处理连接事件。
- Acceptor处理连接事件后，MainReactor将连接分发给具体的SubReactor。
- SubReactor将连接加入连接队列进行监听，并创建对应的handler进行处理各种事件。
- 当事件发生时，SubReactor会调用对应的Handler进行处理。
- Handler通过read读取数据，分发给worke线程池处理。
- worker线程池会分配给具体的线程完成真正的业务处理，并将结果返回给handler。
- handler收到响应后，通过send将处理结果返回给client。
9、讲述一下什么是Netty模型？
工作原理和简单示意图：
- Netty抽象出两个线程组：Boss Group和Worker Group。Boss Group专门负责接收客户端连接；Worker Group专门负责网络的读写。
- Boss Group和Worker Group类型都是NioEventLoopGroup。
- NioEventLoopGroup相当于一个事件循环组，这个组含有多个事件的循环，每一个循环事件是一个NioEventLoop。
- NioEventLoop表示一个不断循环的执行处理任务的线程，每一个NioEventLoop都有一个Selector和TaskQueue，Selector用于监听绑定在其上的socket的网络通迅；TaskQueue是一个任务队列。
- NioEventLoopGroup可以有多个线程，即NioEventLoopGroup可以对应多个NioEventLoop。
- 每一个Boss NioEventLoop 执行步骤：
- step1：轮询accept事件
- step2：处理accept事件，与client建立连接，生成NioSocketChannel，并将其注册到某个 Worker NioEventLoop上的selector。
- step3：处理任务队列，即runAllTasks。
- 每一个Worker NioEventLoop执行步骤：
- step1：轮询read和write事件
- step2：每一个IO事件，即read和write事件，在对应的NioSocketChannel处理。
- step3：处理任务队列，即runAllTasks。
- 每一个Worker NioEventLoop处理业务时，会使用pipeline(管道)，管道维护了很多处理器，处理器可以是netty自带的，也可以自定义。
客户端/服务器端-简单通讯案列：
maven 引入netty依赖：io.netty netty-all 4.1.20.Final compile
Server端：
package com.netty.test01;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;/** * @author 精彩猿笔记 */public class NettyServer { public static void main(String[] args) { //创建线程组BossGroup,处理连接请求 EventLoopGroup bossGroup = new NioEventLoopGroup(1); //创建线程组workerGroup,完成和客户端具体的业务处理 //无参：则workerGroup含有的子线程个数=NettyRuntime.availableProcessors() * 2,即CPU核数*2 EventLoopGroup workerGroup = new NioEventLoopGroup(); try { //创建服务器端启动对,用来配置参数 ServerBootstrap bootstrap = new ServerBootstrap(); //设置参数 //设置两个线程组 bootstrap.group(bossGroup,workerGroup); //使用NioServerSocketChannel作为服务器的通道 bootstrap.channel(NioServerSocketChannel.class); //设置线程队列得到连接个数 bootstrap.option(ChannelOption.SO_BACKLOG,1024); //设置保持活动的连接状态 bootstrap.childOption(ChannelOption.SO_KEEPALIVE,true); bootstrap.childHandler(new ChannelInitializer () { //给pipeLine设置处理器 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { //给workerGroup的EventLoop对应的管道设置处理器 //处理器可以用netty自带的，也可以自己创建处理器，如：NettyServerHandler socketChannel.pipeline().addLast(new NettyServerHandler()); } }); System.out.println("Netty服务端启动...is OK!"); //绑定一个端口并启动服务器,生成一个ChannelFuture对象 绑定：bind ChannelFuture channelFuture = bootstrap.bind(8888).sync(); //对关闭通道进行监听 channelFuture.channel().closeFuture().sync(); }catch (Exception e){ e.printStackTrace(); } finally { //异常~优雅的关闭(netty提供) bossGroup.shutdownGracefully(); workerGroup.shutdownGracefully(); } }} Server端处理器：
package com.netty.test01;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.CharsetUtil;/** * 自定义Handler需要继承netty规定的某个 XXXHandlerAdapter * idea: Ctrl＋O可以选择父类的方法进行重写 */public class NettyServerHandler extends ChannelInboundHandlerAdapter { /** * 当通道有读取事件时，就会触发channelRead * @param ctx：上下文对象，含有：channel、pipeLine * @param msg：客户端发送的实际的数据 * @throws Exception */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //将msg转成一个ByteBuf ByteBuf byteBuf = (ByteBuf) msg; System.out.println("客户端[IP:"+ctx.channel().remoteAddress()+"]说："+byteBuf.toString(CharsetUtil.UTF_8)); } /** * 数据读取完毕时，就会触发channelReadComplete * @param ctx 上下文对象 * @throws Exception */ @Override public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { //将数据写入到缓存，并刷新 writeAndFlush String resMag = "客户端,你好吖!"; //一般需要对发送的数据进行编码 ctx.writeAndFlush(Unpooled.copiedBuffer(resMag,CharsetUtil.UTF_8)); } /** * 异常处理，关闭通道 * @param ctx 上下文对象 * @param cause * @throws Exception */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭 cause.printStackTrace(); ctx.close(); }} Client端：
package com.netty.test01;import io.netty.bootstrap.Bootstrap;import io.netty.channel.ChannelFuture;import io.netty.channel.ChannelInitializer;import io.netty.channel.EventLoopGroup;import io.netty.channel.nio.NioEventLoopGroup;import io.netty.channel.socket.SocketChannel;import io.netty.channel.socket.nio.NioSocketChannel;/** * @author 精彩猿笔记 */public class NettyClient { public static void main(String[] args){ //客户端需要一个事件循环组 EventLoopGroup eventExecutors = new NioEventLoopGroup(); try{ //创建客户端启动对象，客户端使用的是Bootstrap Bootstrap bootstrap = new Bootstrap(); //设置相关参数 //设置线程组 bootstrap.group(eventExecutors); //设置通道处理类 bootstrap.channel(NioSocketChannel.class); bootstrap.handler(new ChannelInitializer () { //设置处理器 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { socketChannel.pipeline().addLast(new NettyClientHandler()); } }); System.out.println("Netty客户端启动...is OK!"); //启动客户端去连接服务器端 连接：connect ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 8888).sync(); //对关闭通道进行监听 channelFuture.channel().closeFuture().sync(); }catch (Exception e){ e.printStackTrace(); } finally { //异常~优雅的关闭(netty提供) eventExecutors.shutdownGracefully(); } }} Client端处理器：
package com.netty.test01;import io.netty.buffer.ByteBuf;import io.netty.buffer.Unpooled;import io.netty.channel.ChannelHandlerContext;import io.netty.channel.ChannelInboundHandlerAdapter;import io.netty.util.CharsetUtil;public class NettyClientHandler extends ChannelInboundHandlerAdapter { /** * 当通道就绪就会触发该方法 * @param ctx 上下文对象 * @throws Exception */ @Override public void channelActive(ChannelHandlerContext ctx) throws Exception { //将数据写入到缓存，并刷新 writeAndFlush String resMag = "服务端,你好吖!"; //一般需要对发送的数据进行编码 ctx.writeAndFlush(Unpooled.copiedBuffer(resMag,CharsetUtil.UTF_8)); } /** * 当通道有读取事件时，就会触发channelRead * @param ctx：上下文对象，含有：channel、pipeLine * @param msg：客户端发送的实际的数据 * @throws Exception */ @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { //将msg转成一个ByteBuf ByteBuf byteBuf = (ByteBuf) msg; System.out.println("服务端[IP:"+ctx.channel().remoteAddress()+"]说："+byteBuf.toString(CharsetUtil.UTF_8)); } /** * 异常处理，关闭通道 * @param ctx 上下文对象 * @param cause * @throws Exception */ @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { //关闭 cause.printStackTrace(); ctx.close(); }} 客户端运行结果：
Netty客户端启动...is OK!服务端[IP:/127.0.0.1:8888]说：客户端,你好吖!
服务器端运行结果：
Netty服务端启动...is OK!客户端[IP:/127.0.0.1:65403]说：服务端,你好吖!
10、任务队列的使用场景有哪些？
当某些业务场景处理时间很长，可能导致客户端超时或者阻塞，则可以将任务加入到任务队列里，可以实现异步处理。
加入任务队列的方式：
- 【方案1】：将一个普通任务加入到TaskQueue中。
- 【方案2】：用户自定义定时任务，将任务加入到scheduledTaskQueue中。
【方案1】和【方案2】都有弊端：即当前处理器handler线程与普通任务的线程是同一个线程，当普通任务线程处理业务逻辑时，当前是阻塞的。因为整个是同一个线程。即如下代码：channelReadThreadId=taskThreadId=scheduleThreadId；channelReadThreadName=taskThreadName=scheduleThreadName。
【方案1】和【方案2】代码简单实现如下：
@Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("channelReadThreadId="+Thread.currentThread().getId()); System.out.println("channelReadThreadName="+Thread.currentThread().getName()); //***************方案1：将一个普通任务加入到TaskQueue中 ctx.channel().eventLoop().execute(new Runnable() { public void run() { try { //通过让线程休眠 模拟业务处理很长时间的场景 Thread.sleep(2*1000); System.out.println("taskThreadId="+Thread.currentThread().getId()); System.out.println("scheduleThreadName="+Thread.currentThread().getName()); } catch (Exception e) { e.printStackTrace(); } } }); //***************方案2：用户自定义定时任务,将任务加入到scheduledTaskQueue,10秒后执行 ctx.channel().eventLoop().schedule(new Runnable() { public void run() { try { //通过让线程休眠 模拟业务处理很长时间的场景 Thread.sleep(3*1000); System.out.println("scheduleThreadId="+Thread.currentThread().getId()); System.out.println("scheduleThreadName="+Thread.currentThread().getName()); } catch (Exception e) { e.printStackTrace(); } } },10, TimeUnit.SECONDS); //将msg转成一个ByteBuf ByteBuf byteBuf = (ByteBuf) msg; System.out.println("客户端[IP:"+ctx.channel().remoteAddress()+"]说："+byteBuf.toString(CharsetUtil.UTF_8)); }输出结果：channelReadThreadId=15channelReadThreadName=nioEventLoopGroup-3-1客户端[IP:/127.0.0.1:56418]说：服务端,你好吖!taskThreadId=15scheduleThreadName=nioEventLoopGroup-3-1scheduleThreadId=15scheduleThreadName=nioEventLoopGroup-3-1 更好的解决方案是：将耗时任务添加到异步线程池中。
-
【方案3】：在Handler中加入业务线程池。【如下Handler线程与耗时任务线程非同一个线程，耗时任务执行完后，在执行pipeline write方法的时候，会将这个任务交给IO线程】【使用灵活，单异步会拖长接口的响应时间，可能导致在规定的响应时间内未响应。】
public class NettyServerHandler extends ChannelInboundHandlerAdapter { /**在handler中自定义业务线程池,这里在线程池中创建了8个线程*/ EventExecutorGroup executorGroup= new DefaultEventExecutorGroup(8); @Override public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { System.out.println("channelReadThreadId="+Thread.currentThread().getId()); System.out.println("channelReadThreadName="+Thread.currentThread().getName()); //将耗时任务1 放入线程池中 executorGroup.submit(new Callable -
【方案4】：Context中添加线程池。【是Netty的标准方式，将整个Handler交给业务线程池，不够灵活】
public class NettyServer { /**自定义业务线程池,这里在线程池中创建了2个子线程*/ static final EventExecutorGroup executorGroup = new DefaultEventExecutorGroup(2); ............ //给workerGroup添加处理器 bootstrap.childHandler(new ChannelInitializer() { //给pipeLine设置处理器 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { //给workerGroup的EventLoop对应的管道设置处理器 //处理器可以用netty自带的，也可以自己创建处理器，如：NettyServerHandler //socketChannel.pipeline().addLast(new NettyServerHandler()); //将NettyServerHandler加入到EventExecutorGroup线程池中 socketChannel.pipeline().addLast(executorGroup,new NettyServerHandler()); } });}
11、说一下什么是Netty异步模型的Futrue-Listener机制？
Netty的异步模型是建立在future和callback之上的。Future的核心思想是：在调用方法后会立马返回一个Future，后续可以通过Future去监听这个方法的执行过程(即：Futrue-Listener机制)。
当Future对象刚刚创建时，处理非完成状态，调用者可以通过返回的ChannelFuture来获取操作的执行状态，注册监听方法执行完成后的操作。 常见操作：- 通过isDone 方法来判断当前操作是否完成
- 通过isSuccess 方法来判断已完成的操作是否成功
- 通过getCause 方法来回去已完成的操作失败的原因
- 通过isCancelled 方法来判断已完成的当前操作是否被取消
- 通过 addListener 方法来注册监听器
- ……等等
12、说说Bootstrap和ServerBootstrap分别做什么用的？
Bootstrap是引导的意思，主要负责配置整个Netty程序，串联各个组件。Bootstrap类是客户端程序启动引导类；ServerBootstrap是服务器端启动引导类。
常见的配置方法有：
- public B group(EventLoopGroup group)：是Bootstrap用于客户端，用来设置一个EventLoopGroup。
- public ServerBootstrap group(EventLoopGroup parentGroup, EventLoopGroup childGroup)：是ServerBootstrap用于服务器端，用来设置两个EventLoopGroup。
- public B channel(Class<? extends C> channelClass)：是Bootstrap和ServerBootstrap用来设置一个通道的实现类。
- public B option(ChannelOption option, T value)：是Bootstrap和ServerBootstrap用来给服务器端 Channel添加配置。
- public ServerBootstrap childOption(ChannelOption childOption, T value)：是ServerBootstrap用来给接收的通道添加配置。
- public B handler(ChannelHandler handler)：是Bootstrap和ServerBootstrap用来给Boss Group添加处理器。
- public ServerBootstrap childHandler(ChannelHandler childHandler)：是ServerBootstrap用来给Worker Group添加处理器。
- public ChannelFuture bind(int inetPort)：是ServerBootstrap用于服务器端，用来设置占用的端口号。
- public ChannelFuture connect(String inetHost, int inetPort)：是Bootstrap用于客户端，用来连接服务器。
- ……等等
13、说说Netty中的Channel有哪些常用的类型？
Channel是Netty通信的组件，不同协议、不同阻塞类型的连接都有不同的Channel与之对应，常用的Channel类型：
- NioSocketChannel：异步的客户端TCP Socket连接
- NioServerSocketChannel：异步的服务器端TCP Socket连接
- NioDatagramChannel：异步的UDP连接
- NioSctpChannel：异步的客户端Sctp连接
- NioSctpServerChannel：异步的服务器端Sctp连接
- ……等等
14、说说什么是ChannelHandler，及其重要子类和重要的基于事件监听的方法有哪些？
ChannelHandler 是一个接口，处理I/O事件或者连接I/O操作，将其转发到其ChannelPipeline(业务处理链)中的下一个个处理程序。
ChannelHandler本身没有提供很多方法，但是其子类非常多提供里很多重要的基于事件监听的方法。 主要的子类关系图： 主要的基于事件监听的方法： //一旦建立连接就会触发该事件，ChannelHandler接口定义的方法 public void handlerAdded(ChannelHandlerContext ctx) throws Exception { } //一旦断开连接时会触发，ChannelHandler接口定义的方法 public void handlerRemoved(ChannelHandlerContext ctx) throws Exception { } //通道被注册时发生事件 public void channelRegistered(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelRegistered(); } //通道被注销时发生事件 public void channelUnregistered(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelUnregistered(); } //通道就绪事件，活动状态 public void channelActive(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelActive(); } //通道非活动状态 public void channelInactive(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelInactive(); } //通道读取数据事件 public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { ctx.fireChannelRead(msg); } //通道读取数据完毕后的事件 public void channelReadComplete(ChannelHandlerContext ctx) throws Exception { ctx.fireChannelReadComplete(); } //通道异常事件 public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { ctx.fireExceptionCaught(cause); } ........等等 15、说一下Netty中的ChannelPipeline是什么？
ChannelPipeline是Handler的集合，它负责处理和拦截入栈(inBound)或出栈(outBound)的事件和操作。
在Netty中每一个Channel都有一个ChannelPipeline与之对应，一个ChannelPipeline维护了一个由ChannelHandlerContext组成的双向链表。 入栈事件和出栈事件在一个双向链表中，入栈事件会从链表的head节点(头节点)往后传递到最后一个Handler；出栈事件会从链表tail节点(尾结点)往前传递到最前一个出栈的Handler；两种类型的Handler互不干扰。 关系图如下： 常用的方法： - ChannelPipeline addFirst(ChannelHandler… var1)：把业务处理类(handler)加入到链表的第一个位置。
- ChannelPipeline addLast(ChannelHandler… var1)：把业务处理类(handler)加入到链表的最后一个位置。
- ……等等更多重载的方法。
16、说一下Netty中的ChannelHandlerContext是什么？
ChannelHandlerContext 保存了Channel相关的所有上下文信息，同时关联一个ChannelHandler对象，即ChannelHandlerContext中包含一个具体的事件处理器ChannelHandler，同时也绑定了对应的PipeLine和Channel的信息。
常用的方法：- ChannelFuture close()：关闭通道
- ChannelOutboundInvoker flush()：刷新
- ChannelFuture writeAndFlush(Object var1)：将数据写入ChannelPipeline中当前ChannelHandler的下一个ChannelHandler开始处理。
- ……等等
17、说一下Netty中的ChannelOption是什么？
在Netty创建Channel实例后，一般需要设置ChannelOption参数。
常见的设置如下：- ChannelOption.SO_BACKLOG：对应TCP/IP协议listen函数中的backlog参数(backlog指定队列的大小)，用来初始化服务器可连接队列大小。【服务端处理客户端连接请求是顺序处理的，所以同一时间只能处理一个客户端连接，多个客户端来的时候，服务端将不能处理的客户端连接请求放在队列中等待处理，backlog参数指定了队列的大小。】
- ChannelOption.SO_REUSEADDR：允许重复使用本地地址和端口。【某个服务器进程占用了TCP的80端口进行监听，此时再次监听该端口就会返回错误，使用该参数就可以解决问题，该参数允许共用该端口。】
- ChannelOption.SO_KEEPALIVE：一直保持连接活动状态。【当设置该选项以后，如果在两小时内没有数据的通信时，TCP会自动发送一个活动探测数据报文。】
- ChannelOption.SO_SNDBUF：用于操作发送缓冲区大小。【接收缓冲区用于保存网络协议站内收到的数据，直到应用程序读取成功。】
- ChannelOption.SO_RCVBUF：用于接受缓冲区大小。【发送缓冲区用于保存发送数据，直到发送成功。】
- ChannelOption.SO_LINGER：阻塞close()的调用时间，直到数据完全发送。【调用close（）方法的时候，函数返回，在可能的情况下，尽量发送数据，不一定保证会发送剩余的数据，造成了数据的不确定性。】
- ChannelOption.TCP_NODELAY：是禁止使用Nagle算法。【Nagle算法是将小的数据包组装为更大的帧然后进行发送，而不是输入一次发送一次,因此在数据包不足的时候会等待其他数据的到了，组装成大的数据包进行发送，虽然该方式有效提高网络的有效负载，但是却造成了延时，及TCP粘包和拆包问题。】
18、说一下Netty中的Unpooled类是什么？
Unpooled 是Netty提供的一个专门用来操作缓冲区(Netty的数据容器)的工具类。
常用的方法：- public static ByteBuf buffer(int initialCapacity)
- public static ByteBuf directBuffer(int initialCapacity)
- public static ByteBuf wrappedBuffer(byte[] array)
- public static ByteBuf copiedBuffer(ByteBuf buffer)
- ……等等
19、Netty的ByteBuf与NIO中的ByteBuffer有什么区别？
Netty的数据容器ByteBuf。
三个重要的属性： 因为Netty的ByteBuf维护了writerIndex和readerIndex，所以读写之前切换不需要使用到flip()，这也和NIO有区别的。- readerIndex：下一个可读数据下标，读取方法比如：ByteBuf.read+基本类型()，说明getByte()也能读取数据，但是不会让readerIndex值改变。
- writerIndex：下一个可写数据下边，写方法比如：ByteBuf.write+基本类型()
- capacity：缓冲区容量
属性之间的区域描述：
- 0~readerIndex：已读数据区域
- readerIndex~writerIndex：可读数据区域
- writerIndex~capacity：可写数据区域
- public abstract int capacity()：获取capacity
- public abstract int readerIndex()：获取readerIndex
- public abstract int writerIndex()：获取writerIndex
- public abstract int readableBytes()：获取可读的字节数
- public abstract int writableBytes()：获取可写的字节数
- public abstract ByteBuf clear()：清空
- public abstract byte[] array()：获取整个缓冲区的内容
- public abstract CharSequence getCharSequence(int var1, int var2, Charset var3)：读取从var1开始后的var2个字节，用var3编码格式
- ……等等。
20、聊聊Netty提供了那些编码/解码器，存在那些问题？
codec(编/解码器)由两个组成部分有两个：
- encoder(编码器)：将业务数据转换成字节码数据
- decoder(解码器)：将字节码数据转换成业务数据
Netty发送和接收一个消息时都会产生一次数据转换：
- 入栈消息会被解码：即可以理解为接收消息端，当接收消息(入栈)时，需要对消息进行解码。
- 出栈消息会被编码：即可以理解为发送消息端，当发送消息(出栈)时，需要对消息进行编码。
Netty提供了一系列实用的编码、解码器。它们都实现了ChannelInboundHandler(入栈解码：对应decode()方法) 或者ChannelOutboundHandler(出栈编码：对应encode()方法) 。
简答理解图如下：
自定义编码器-简单例子：
/** * 自定义编码器 MessageToByteEncoder extends ChannelOutboundHandlerAdapter * @author 精彩猿笔记 */public class MyMessageToByteEncoder extends MessageToByteEncoder{ /**自定义编码器：重写父类的编码encode方法*/ @Override protected void encode(ChannelHandlerContext channelHandlerContext, Integer integer, ByteBuf byteBuf) throws Exception { //以int格式进行编码 byteBuf.writeInt(integer); }}
当自定义编码器(重写父类的encode()方法)时，需要编码的消息类型必须与自定义待处理的消息类型一致，否则不会执行自定义handler内的业务逻辑，直接将需要编码的数据写入，没有达到自定义编码的业务。
例如：JDK 1.8【MessageToByteEncoder抽象类的write方法】源码如下：public abstract class MessageToByteEncoder extends ChannelOutboundHandlerAdapter { public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception { ByteBuf buf = null; try { //*************this.acceptOutboundMessage(msg) 判断需要编码的消息类型是否与自定义待处理的消息类型一致 if (this.acceptOutboundMessage(msg)) { I cast = msg; buf = this.allocateBuffer(ctx, msg, this.preferDirect); try { //*************这调用自定义重写的encode方法，完成编码业务逻辑 this.encode(ctx, cast, buf); } finally { ReferenceCountUtil.release(msg); } if (buf.isReadable()) { ctx.write(buf, promise); } else { buf.release(); ctx.write(Unpooled.EMPTY_BUFFER, promise); } buf = null; } else { //*************不一致，则不会处理自定义编码业务逻辑,直接将需要编码的数据写入，没有达到自定义编码的业务。 ctx.write(msg, promise); } } catch (EncoderException var17) { throw var17; } catch (Throwable var18) { throw new EncoderException(var18); } finally { if (buf != null) { buf.release(); } } }.......} 自定义解码器-简单例子：
/** * 自定义解码器 ByteToMessageDecoder extends ChannelInboundHandlerAdapter * @author 精彩猿笔记 */public class MyByteToMessageDecoder extends ByteToMessageDecoder { /**自定义解码器：重写父类的编码decode方法*/ @Override protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List Netty自身提供了一些常用的codec(编/解码器)
- Netty encoder自带编码器如：
- StringEncoder：对字符串数据进行编码
- ObjectEncoder：对java对象进行编码
- ZlibEncoder：将压缩数据编码
- HttpObjectEncoder：一个http数据编码
- ……等等
- Netty decoder自带解码器如：
- StringDecoder：对字符串数据进行解码
- ObjectDecoder：对java对象进行解码
- ZlibDecoder：将压缩数据解码
- HttpObjectDecoder：一个http数据解码
- LineBasedFrameDecoder：它使用行尾控制字符(\n或者\r\n)作为分隔符来解析数据
- DelimiterBasedFrameDecoder：使用自定义的特殊字符作为消息的分隔符。
- LengthFieldBasedFrameDecoder：通过指定长度来标识整包数据。
- ……等等
Netty自带的编码、解码底层使用的仍然是Java序列化技术，而Java序列化技术本身效率不是很高，所以存在如下问题：
- 无法跨语言，即编码端用什么语言，解码端就也得用什么语言
- 序列化后体积太大，是二进制编码的5倍多
- 序列化性能低
解决方案：可以推荐使用Google的Protobuf。
21、你对Google的Protobuf的了解有多少？
Protobuf(Google Protocol Buffers) 是Google发布的开源项目。是一种轻便高效的结构化存储格式，可以用于结构化数据串行化，或者说序列化。很适合做数据存储 和RPC[远程过程调用 remote procedure call]。
特点:
支持跨平台、跨语言(即发送端和接收端可以是不同的编程语言，目前支持绝大多数语言，比如C++、C#、Java、python等等)、高性能、高可靠。 编译器：
ProtoBuf是以massage的方式来管理数据的，Protobuf是以 .proto 结尾 的文件，通过protoc.exe编译器 根据.proto自动生成使用的编程语言对应的代码。22、Netty对TCP粘包拆包问题提出了那些解决方案？
简单图解分析，什么是TCP粘包拆包：
Netty使用自定义协议+编码解码 来解决Tcp粘包和拆包问题。关键就是要解决服务器每次读取数据长度的问题 ，这个解决就不会出现多读或者少读的问题，从而避免了TCP粘包拆包问题。 核心代码： 
23、源码分析Boss Group和Worker Group 对象的创建过程？
例：EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(); 两个对象是整个Netty的核心对象。bossGroup 用于接收TCP请求，它会将请求交给workerGroup，workerGroup会获取真正的连接，然后和连接进行通信。EventLoopGroup是一个事件循环线程组，含有多个EventLoop。new NioEventLoopGroup(1); 有参构造，这个1表示生成一个子线程的bossGroup。
new NioEventLoopGroup(); 无参构造，则workerGroup含有的子线程个数，即CPU核数*2。JDK 1.8 源码如下：private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) { super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);} 底层都会创建EventExecutor数组：
JDK 1.8 源码如下：if (executor == null) { executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory());}this.children = new EventExecutor[nThreads]; 24、源码分析EventLoopGroup创建的过程？
JDK 1.8 源码如下：/***@param nThreads 使用线程数 默认为CPU核数*2*@param executor 执行器，如果传null则采用Netty默认的线程工厂和默认的执行器ThreadPerTaskExecutor*@param chooserFactory 单例 new DefaultEventExecutorChooserFactory()*@param args args在创建执行器的时候穿日固定参数*/protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { this.terminatedChildren = new AtomicInteger(); this.terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE); if (nThreads <= 0) { throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); } else { //如果传null则采用Netty默认的线程工厂和默认的执行器ThreadPerTaskExecutor if (executor == null) { executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); } //创建指定线程数的执行器数组 this.children = new EventExecutor[nThreads]; int j; for(int i = 0; i < nThreads; ++i) { boolean success = false; boolean var18 = false; try { var18 = true; //创建 new NIOEventLoop this.children[i] = this.newChild((Executor)executor, args); success = true; var18 = false; } catch (Exception var19) { throw new IllegalStateException("failed to create a child event loop", var19); } finally { if (var18) { if (!success) { int j; for(j = 0; j < i; ++j) { this.children[j].shutdownGracefully(); } for(j = 0; j < i; ++j) { EventExecutor e = this.children[j]; try { while(!e.isTerminated()) { e.awaitTermination(2147483647L, TimeUnit.SECONDS); } } catch (InterruptedException var20) { Thread.currentThread().interrupt(); break; } } } } } if (!success) { for(j = 0; j < i; ++j) { //优雅的关闭 this.children[j].shutdownGracefully(); } for(j = 0; j < i; ++j) { EventExecutor e = this.children[j]; try { while(!e.isTerminated()) { e.awaitTermination(2147483647L, TimeUnit.SECONDS); } } catch (InterruptedException var22) { Thread.currentThread().interrupt(); break; } } } } this.chooser = chooserFactory.newChooser(this.children); FutureListener 25、通过源码对NioEventLoop三个步骤的验证？
// JDK 1.8 NioEventLoop.run()如下：protected void run() { while(true) { while(true) { try { switch(this.selectStrategy.calculateStrategy(this.selectNowSupplier, this.hasTasks())) { case -2: continue; case -1: //******************************step1:select //调用selector的select方法，默认阻塞一秒钟，如果有定时任务，则在定时任务剩余时间的基础上在家0.5秒进行阻塞，当执行execute方法添加任务的时候，唤醒selector，防止selector阻塞时间过长。 this.select(this.wakenUp.getAndSet(false)); if (this.wakenUp.get()) { this.selector.wakeup(); } default: this.cancelledKeys = 0; this.needsToSelectAgain = false; int ioRatio = this.ioRatio; if (ioRatio == 100) { try { //******************************step2:processSelectedKeys //调用processSelectedKeys方法对selectKey进行处理 this.processSelectedKeys(); } finally { //******************************step3:runAllTasks //按照ioRatio 的比例执行runAllTasks方法，默认IO任务时间和非IO任务时间是相同的，你也可以根据你的应用特点进行调优 //比如：非IO任务较多，那么你就将ioRatio 调小一点，这样非IO任务就能执行的长一点，房子队列积攒过多的任务。 this.runAllTasks(); } } else { long ioStartTime = System.nanoTime(); boolean var13 = false; try { var13 = true; this.processSelectedKeys(); var13 = false; } finally { if (var13) { long ioTime = System.nanoTime() - ioStartTime; this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio); } } long ioTime = System.nanoTime() - ioStartTime; //******************************step3:runAllTasks this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio); } } } catch (Throwable var21) { handleLoopException(var21); } try { if (this.isShuttingDown()) { this.closeAll(); if (this.confirmShutdown()) { return; } } } catch (Throwable var18) { handleLoopException(var18); } } }} 26、源码分析ServerBootstrap配置过程？
根据如下例子说明：
//创建服务器端启动对,用来配置参数ServerBootstrap bootstrap = new ServerBootstrap();//设置参数//设置两个线程组bootstrap.group(bossGroup,workerGroup);//使用NioServerSocketChannel作为服务器的通道，应道类将通过这个Class对象反射创建ChannelFactorybootstrap.channel(NioServerSocketChannel.class);//设置线程队列得到连接个数，放入private final Map, Object> options = new LinkedHashMap(); 进行管理bootstrap.option(ChannelOption.SO_BACKLOG,1024);//设置保持活动的连接状态bootstrap.childOption(ChannelOption.SO_KEEPALIVE,true);//给bossGroup添加日志处理器，传入一个handler，这个handler只属于ServerSocketChannel，而不属于SocketChannelbootstrap.handler(new LoggingHandler(LogLevel.ERROR));//给workerGroup添加处理器,传入一个handler，这个handler将会在每个客户端连接的时候调用，供socketChannel使用bootstrap.childHandler(new ChannelInitializer () { //给pipeLine设置处理器 @Override protected void initChannel(SocketChannel socketChannel) throws Exception { //给workerGroup的EventLoop对应的管道设置处理器 //处理器可以用netty自带的，也可以自己创建处理器，如：NettyServerHandler socketChannel.pipeline().addLast(new NettyServerHandler()); } });
27、对源码绑定端口的分析？
例： ChannelFuture channelFuture = bootstrap.bind(8888).sync(); //服务器就在通过ServerBootstrap对象的bind方法来绑定一个端口的，生成一个ChannelFuture对象。
通过bind调用到底层【private ChannelFuture doBind(final SocketAddress localAddress) 】方法。//JDK 1.8 源码如下：private ChannelFuture doBind(final SocketAddress localAddress) { //**************initAndRegister final ChannelFuture regFuture = this.initAndRegister(); final Channel channel = regFuture.channel(); if (regFuture.cause() != null) { return regFuture; } else if (regFuture.isDone()) { ChannelPromise promise = channel.newPromise(); //**************doBind0 doBind0(regFuture, channel, localAddress, promise); return promise; } ........ doBind方法里有两个重要的方法，分别是：
-
this.initAndRegister()，如下：
//JDK 1.8 源码如下：final ChannelFuture initAndRegister() { Channel channel = null; try { /**主要完成： * 1.通过NIO的SelectorProvider的openServerChannel的方法得到JDK的channel，目的是通过Nett包装JDK的channel。 * 2.创建一个ChannelId；创建了一个NIOMessageUnsafe，用于操作下消息；穿件了一个DefaultChannelPipeline管道，是双向链表，用于过滤所有进出的消息 * 穿件一个NioServerSocketChannelConfig对象，用于对外展示一些配置。 */ channel = this.channelFactory.newChannel(); /** *主要完成： * 1.设置NioServerSocketChannel的TCP属性 * 2.由于LinkedHashMap是非线程安全的，所以使用了同步进行处理 * 3.对NioServerSocketChannel的ChannelPipeline添加ChannelInitializer处理器。 */ this.init(channel); } ........} -
doBind0(regFuture, channel, localAddress, promise)；
- channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
- unsafe.bind(localAddress, promise);
- protected void doBind(SocketAddress localAddress) throws Exception 【NioServerSocketChannel类】
- unsafe.bind(localAddress, promise);
- channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
28、请结合Netty源码说一下Netty接收请求过程？
总体流程：接收连接 -----> 创建一个新的NioSocketChannel -----> 注册一个Worker EventLoop上 -----> 注册select Read事件。
- 服务器轮询Accept事件，获取事件后调用unsafe的read方法，这个unsafe是serverSocket的内部类，该方法内部由两部分组成。
- doReadMessage用于创建NioSocketChannel对象，该对象包装JDK的NIO channel客户端。该方法会创建ServerSocketChannel类似创建相关的pipeline、unsafe、config。
- 随后执行pipeline.fireChannelRead方法，并将自己绑定到一个选择器选择的workerGroup中的一个EventGroup。并且注册一个0，表示注册成功，但并没有注册读(1)事件。
29、说说ChannelPipeline、ChannelHandler、ChannelHandlerContext之间的关系和创建过程？
ChannelPipeline、ChannelHandler、ChannelHandlerContext之间的关系：
- 每当ServerSocket创建一个新的连接，就会创建一个Socket，对应的就是目标客户端。
- 每一个新的Socket都会分配一个新的ChannelPipeline。
- 每一个ChannelPipeline内部含有多个ChannelHandlerContext。
- 它们组成了双向链表，这个链表调用addLast方法是添加的ChannelHandler节点。
每当创建channelSocket的时候都会创建绑定的Pipeline，一一对应关系，创建Pipeline的时候会创建head节点和tail节点，形成最初的链表。head是实现inBound类型和outBound类型的Handler；tail是实现inBound类型的Handler。在调用pipeline的addLast方法的时候，会根据给定的Handler创建一个Context，然后将这个Context插入到链表的尾端(tail节点前)。
30、源码剖析ChannelPipeline调度Handler过程？
Context包装handler，多个Context在Pipeline中形成了双向链表。入栈叫inBound，有head节点开始；出栈叫outBound，由tail节点来时。
而节点中间的传递通过AbstractChannelHandlerContext类内部的fire系列方法，找到当前节点下一个节点不断的循环传递。是一个过滤器模式完成对Handler的调度。调度过程如下图：
说明：- pipeline首先会调用Context的静态方法fireXXX，并传入Context。
- 然后fireXXX静态方法会调用Context的invoker方法，而invoker方法内部会调用改Context包含的Handler的真正方法 XXX 方法，调用结束后没如果还需要向后传递，就调用Context的fireXXX2方法，循环处理。
31、请结合源码对Netty心跳检测机制的剖析？
Netty作为一个网络应用框架，提供了很多功能，如非常重要的心跳机制heartBeat。通过心跳检查与对方的连接是否有效。这也是RPC远程调用框架必不可少的功能。
Nett提供了IdleStateHandler、ReadTimeoutHandler、writeTimeoutHandler三个Handler检测连接的有效性。-
IdleStateHandler：当连接的空闲时间(读或写)太长时，将会触犯一个IdleStateEvent事件。
IdleStateHandler四个常用属性：- private final boolean observeOutput;//是否考虑出栈时较慢的情况。默认是false-不考虑
- private final long readerIdleTimeNanos;//读事件空闲时间(纳秒，一秒的十亿分之一)
- private final long writerIdleTimeNanos;//写事件空闲时间(纳秒，一秒的十亿分之一)
- private final long allIdleTimeNanos;//读或者写的最大事件空闲时间(纳秒，一秒的十亿分之一)
IdleStateHandler可以可以实现心跳功能，当服务器和客户端在规定时间内没有发生读写事件时，这会触发用户Handler的userEventTriggered方法。用户可以在这个方法中尝试向对方发送信息，如果发送失败，则关闭连接。
IdleStateHandler的实现是基于EventLoop的定时任务，每次读写都会记录一个值，在定会任务运行的时候，会通过计算当前时间和设置时间和上一次时间放生的时间结果，来判断是否空闲。 内部由3个定时任务，分别是：读事件、写事件、读或写事件。 -
ReadTimeoutHandler：继承IdleStateHandler，当触发读空闲事件的时候，就会触发crx.fireExceptionCaught方法，并传入一个ReadTimeoutHandler，然后关闭socket。
-
WriteTimeoutHandler：继承ChannelOutboundHandlerAdapter，当调用write方法的时候，会创建一个定时任务，任务内容是根据传入的promise的完成情况来判断是否超出写的时间。当定时任务根据指定时间开始运行，发现promise的isDone(isDone-判断当前操作是否完成)方法返回false，表明还没写完，说明超时了，则抛出异常。
32、说说RPC的调用过程？
RPC[Remote Procedure Call 远程过程调用] ：是一个计算机通信协议，该协议允许运行在一台计算器的程序调用另一台计算机的程序。两个或多个应用程序都分布在不同的服务器上。
使用场景： 阿里的Dubbo、Google的gRPC、Go语言的rpcx、Apache的thrift、Spring Cloud 等等。 后续更多关于Netty的相关内容会不断更新中……
······
帮助他人，快乐自己，最后，感谢您的阅读！ 所以如有纰漏或者建议，还请读者朋友们在评论区不吝指出！…知识是一种宝贵的资源和财富，益发掘，更益分享…
发表评论
最新留言
白红宇的个人博客 - 记录点点滴滴的事 - 您是第 461278121 位访客