当前位置：首页 > news >正文

Netty源码系列之 EventLoop run()方法源码

news 来源：原创 2024/5/20 9:06:57

EventLoop[实现类为NioEventLoop，我们研究NioEventLoop即可]

EventLoop是一个单线程的线程池

核心作用：处理执行IO操作（accept，read，write事件），普通任务，定时任务

EventLoop封装了Selector复用器，Thread线程，以及任务队列

为什么EventLoop需要一个任务队列？

因为EventLoop是一个单线程的线程池，如果有多个任务请求过来时，可能同时处理不了这么多，所以需要暂时存储到一个任务队列中逐个慢慢处理。

我们上面说了EventLoop封装了Selector复用器，Thread线程，以及任务队列，如何证明？

先看EventLoop体系图，它肯定继承了各种类，各种父类的属性为EventLoop提供了以上属性或者是 EventLoop自己这个类也会封装一些属性。

当然，由于EventLoop只是一个接口，所以当我们需要研究观察NioEventLoop的体系图：

如下分析：

NioEventLoop继承ScheduledExecutorService接口，所以该接口实现类就是NioEventLoop的父类

SingleThreadEventExecutor这个父类提供任务队列和线程这两个属性

3.AbstractScheduledEventExecutor类是NioEventLoop的父类，所以NioEventLoop继承拥有父类的所有属性

AbstractScheduledEventExecutor类为NioEventLoop提供用于存储定时任务的优先级队列这个属性

总结：

NioEventLoop实现了EventLoop接口，NioEventLoop是一个集大成者，它不仅仅是一个普通的线程对象，而是一个具有Selector，任务队列缓存功能等的优秀组件。

由此也可以看出，一个EventLoop[NioEventLoop]对应一个Selector复用器类

所以：EventLoop接口实现类NioEventLoop拥有以下核心属性：

1.private Selector selector; //包装后的Selector对象

2.private Selector unwrappedSelector;//未包装的Selector对象

3.private final Queue<Runnable> taskQueue;//负责存储未执行完的普通任务。因为NioEventLoop是单线程，如果有多个任务，那么任务队列是必须的。

4.private volatile Thread thread;//线程对象，真正执行代码逻辑的对象

5. PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue;//用于存储定时任务的优先级队列。该队列具体底层的实现对应的数据结构HashWheelTimer

这么多属性共同铸就了NioEventLoop的功能特性

问题：NioEventLoop中的Selector选择器什么时候被创建的？

在NioEventLoop的构造器中：

1.final SelectorTuple selectorTuple = openSelector();

该方法就是先获取windows操作系统(或其他操作系统)所对应的最原生的Selector对象，也就是未包装的Selector对象：unwrappedSelector。然后再通过反射技术把unwrappedSelector对应的属性等赋值给Selector，返回SelectorTuple对象，该对象包含Selector对象和unwrappedSelector对象

2.this.selector = selectorTuple.selector;

获取到包装后的Selector对象赋值给属性selector

3.this.unwrappedSelector = selectorTuple.unwrappedSelector;

获取到未包装的Selector对象给属性unwrappedSelector

进入openSelector方法：

openSelector方法：

1.把unwrappedSelector对应的属性赋值给selector

2.会把通过unwappedSelector拿到 selectedKey对应的HashSet集合，然后把该集合中的数据赋值给selector的SelectionKey[] keys;数组里。Netty底层反射完成。

补充：

Selector属性和unwrappedSelector属性二者所对应的对象有什么区别？

在Netty中，这两个属性的区别如下：

private Selector selector: 这是Netty中的一个属性，表示经过包装的Selector对象。Netty在创建和管理Selector时，会对其进行包装，以提供更高级的功能和对底层Selector的优化。这个selector属性是Netty的EventLoop中使用的，用于处理网络事件。通过selector，Netty可以监听和处理多个Channel上的事件，并通过事件驱动的方式进行网络编程。
private Selector unwrappedSelector: 这也是Netty中的一个属性，表示未经包装的原始底层Selector对象。这个unwrappedSelector属性是Netty的EventLoop中使用的，用于处理底层的I/O事件。Netty会使用自己的EventLoopGroup来创建和管理Selector，并将其包装成unwrappedSelector。通过unwrappedSelector，Netty可以对底层Selector进行更高级的操作和管理，例如处理空轮询、优化事件的触发和取消等。

总结起来，selector是经过Netty包装的Selector对象，用于处理网络事件，而unwrappedSelector是未经包装的底层Selector对象，用于处理底层的I/O事件。它们在Netty的EventLoop中扮演不同的角色，分别负责处理不同层次的事件。

Netty的Selector对底层的unwrappedSelector进行了封装，以提供更高级的操作和管理。其中包括处理空轮询的情况，优化事件的触发和取消等。

通过封装unwrappedSelector，Netty可以在底层unwrappedSelector的基础上实现一些额外的功能和优化，以提高网络编程的性能和可靠性，最终得出了NioEventLoop中的selector属性。这些功能和优化包括但不限于：

处理空轮询：当底层Selector在轮询时没有任何事件发生时，Netty会进行特殊处理，避免空轮询的问题，从而提高了事件的处理效率。【具体如何处理空轮询的？后续会分析到，使用的就是一个计数器计数，等到一定条件后，重构Selector】
优化事件的触发和取消：Netty可以根据具体的业务需求和网络情况，对事件的触发和取消进行优化。例如，可以通过调整事件的触发条件和取消条件，避免不必要的事件触发和处理，提高网络应用的性能和响应速度。

自己写一个反射案例吧

@Slf4j
public class User {private String name;
}

package com.messi.netty_source_03.Test02;import lombok.extern.slf4j.Slf4j;import java.lang.reflect.Field;@Slf4j
public class TestUser {public static void main(String[] args) throws NoSuchFieldException, IllegalAccessException {User user = new User();Class<? extends User> userClazz = user.getClass();Field name = userClazz.getDeclaredField("name");log.debug("name {} ", name);//打破封装  让外部的其他类也可以访问User类的私有属性namename.setAccessible(true);//user.namename.set(user, "leomessi");log.debug("value is {} ",name.get(user));}
}

测试：

NioEventLoop中的两个属性:selector属性比unwrappedSelector属性，哪一个使用的多？

selector属性比unwrappedSelector属性用的多

这两者有啥区别：

1.unwrappedSelector是原生的操作系统分配的Selector对象，具有很多Bug，也是java原生NIO所使用的Selector复用器。所以NIO许多bug，如：jdk1.7之前的空轮询bug，莫名其妙的bug。

处理空轮询：当底层Selector在轮询时没有任何事件发生时，Netty会进行特殊处理，避免空轮询的问题，从而提高了事件的处理效率。【具体如何处理空轮询的？后续会分析到，使用的就是一个计数器计数，等到一定条件后，重构Selector】
优化事件的触发和取消：Netty可以根据具体的业务需求和网络情况，对事件的触发和取消进行优化。例如，可以通过调整事件的触发条件和取消条件，避免不必要的事件触发和处理，提高网络应用的性能和响应速度。

到目前为止：我们可以发现，Selector做了什么优化？存储Selector上对应的SelectionKey时，原生JavaNIO使用的是Set集合，而Netty使用的是数组。啥是SelectionKey？在上一个小节的最后总结过。

演示如下：

NIO：

Netty：

总结：

这仅仅是Netty对于Selector的优化。后续还有FastThreadLocal，HashWheelTimer的优化。

Netty让SelectedKey存储形式从javaNIO中的Set集合类型，转变成数组类型。数组类型更加有利于提升数据的遍历查找的性能。但是你不可以说仅仅这一个优化，就让Netty成为了高性能框架。你也不能说在高并发场景下，使用数组的绝对优势。还是那句话，这一个优化只是Netty对javaNIO或java jdk原生提供的类型的优化之一，后续还有FastThreadLocal，HashWheelTimer，这两种数据结构也是Netty对jdk中原生数据结构的优化的代表作。正是因为这么多个优化的共同协作下，才让Netty成为了一个异步事件监听回调的高性能框架。

并且你不能说jdk原生的数据结构不好，而是在Netty通信所在的高并发场景下，我们需要选取适应场景所需的数据结构，如果原生数据结构不符合，那么需要做优化定制。

EventLoop[NioEventLoop]的Nio线程什么时候启动呢？&& 如何进行IO操作，普通任务处理，定时任务处理呢？&& Netty如何解决NIO-Selector空转的问题

测试用例

package com.messi.netty_source_03.Test03;import io.netty.channel.EventLoop;
import io.netty.channel.nio.NioEventLoopGroup;import java.util.concurrent.TimeUnit;public class TestEventloop2 {public static void main(String[] args) {NioEventLoopGroup eventLoopGroup = new NioEventLoopGroup();//设置IoRatio//eventLoopGroup.setIoRatio(80);EventLoop eventLoop = eventLoopGroup.next();//一切准备好后，直接开始执行该任务eventLoop.execute(() -> {System.out.println("hello suns");});//定时任务。一切准备好后，也得再过200秒后才会执行该定时任务eventLoop.schedule(() -> {System.out.println("TestEventloop2.main");}, 200, TimeUnit.SECONDS);}
}

debug源码过程：

6.进入doStartThread方法

核心：run方法【涉及IO,普通任务,定时任务时间的分配 && Netty如何解决NIO-Selector空转的问题】

接着上面进行debug，进入run方法：

我们所做的逻辑就是：不要让selector.select()一直阻塞，因为我们还需要处理定时任务和普通任务。

三种情况：

情况1：

IO操作全部做完后再执行普通任务。但是当普通任务执行的过程中，又有新的IO操作来了，如果任务还没有执行完，那么还是会执行未处理完的普通任务【为什么会这样？因为ranTasks=runAllTasks()没有传递参数，所以执行任务操作时，没有限制时间，会一直执行到任务处理结束】。显然，这种ioRatio==100的做法是不正确的。所以说，要设计一个比例，让普通任务不能完全阻塞影响到后续进入的IO网络通信的执行【因为网络通信是不能延迟的！就算普通任务没有执行完，网络通信IO也要先执行，所以说设置一个好的ioRatio比例值是非常重要的】

情况2：

ioRatio!=100，先执行完全部的IO操作。然后执行任务时会根据ioRatio计算出一个执行时间长度限制，即使任务到了时间限制后没有执行完毕，那么也不能再执行任务了，而是会跳转去执行IO操作，IO操作可以无限制时间的去执行，直到IO执行完毕【因为IO操作是网络通信，用户不能等】

这样就解决了情况1的问题【任务无截止时间的执行导致后续进来的IO执行的阻塞】

情况3：

此时只有任务需要执行，没有IO操作。会没有限制时间的执行所有的任务。

如何解决epoll空转问题？

在linux操作系统下的epoll模型，如果我们把java代码部署在linux操作系统上并且会调用epoll的话。

在原生jdk-NIO，从jdk1.6开始出现selector.select()空转[epoll空转]问题，号称在jdk1.7已经修复了，但是jdk1.8还是会存在空转的问题。只不过空转的几率会越来越小。修复了很多版本。但是记住：空转这个问题是一个极其偶发的事情，几率很低。

先说一下，什么叫做epoll空转？

所谓空转就是：当前没有IO事件或任务[普通任务或定时任务]时，本应该阻塞的selector.select()方法突然停止阻塞然后不断的执行while循环，所谓空轮询【空转】就是：没有意义的不断循环，因为此时压根就没有任务或IO事件需要你处理。一旦空轮询发生，也就是一直会while死循环，那么cpu占有率达到100%是迟早的事情

但是Netty解决了这个臭名昭著的问题。

如何解决的呢？就是在run方法中定义了一个计数器变量，当没有IO或任务执行时，但是该计数器变量在一定时间内增加到512后会执行重构selector的操作。重构后的selector可能就不会出现空转问题了，因为空转是一个极其偶发的事情，几率很低。但是有一点你需要知道：Netty并没有从本质上解决jdk原生NIO调用epoll时的空转问题，而是当出现空转时，我们重构selector，毕竟空转就是一个偶发几率低的事件，那么重构后很大可能就不会空转了。

直接看源码中如何解决的：

Selector重构涉及到很多方面，不再过多描述。

EventLoop如何处理IO操作

NioEventLoop找到selector对象，selector找到对应所有的SelectionKey，SelectionKey再找到对应的attachment附件：Channel对象。拿到Channel对象可以获取到pipeline对象，进行网络IO操作。

EventLoop--->NioEventLoop--->selector--->SelectionKey[]

--->NioServerSocketChannel/NioSocketChannel[附件]--->pipeline--->网络IO操作

debug代码如下

package com.messi.netty_source_03.class_04;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;public class NettyServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup eventLoopGroup = new NioEventLoopGroup();ServerBootstrap serverBootstrap = new ServerBootstrap();serverBootstrap.channel(NioServerSocketChannel.class);serverBootstrap.group(eventLoopGroup);serverBootstrap.childHandler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler());}});Channel channel = serverBootstrap.bind(8000).sync().channel();channel.closeFuture().sync();}}

package com.messi.netty_source_03.class_04;import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.logging.LoggingHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;import java.net.InetSocketAddress;
import java.nio.charset.Charset;public class MyNettyClient {private static final Logger log = LoggerFactory.getLogger(MyNettyClient.class);public static void main(String[] args) throws InterruptedException {log.debug("myNettyClientStarter------");EventLoopGroup eventLoopGroup = new NioEventLoopGroup();Bootstrap bootstrap = new Bootstrap();bootstrap.channel(NioSocketChannel.class);Bootstrap group = bootstrap.group(eventLoopGroup);//32 ---> 1 IO操作 31线程bootstrap.handler(new ChannelInitializer<NioSocketChannel>() {@Overrideprotected void initChannel(NioSocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler());ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){@Overridepublic void channelActive(ChannelHandlerContext ctx) throws Exception {if (ctx.channel().isWritable()) {ByteBufAllocator alloc = ctx.alloc();ByteBuf buffer = alloc.buffer();buffer.writeCharSequence("xiaohei", Charset.defaultCharset());ctx.writeAndFlush(buffer);}}});}});Channel channel = bootstrap.connect(new InetSocketAddress(8000)).sync().channel();channel.closeFuture().sync();}
}

debug流程【主要记录IO操作的流程】

预备工作：

debug方式启动服务端

在服务端debug调试完成bind绑定操作后，【bind绑定的流程省略】。

bind流程：完成NioServerSocketChannel的注册，使得服务端完成初始化操作。给NioServerSocketChannel分配一个对应的SelectionKey。并且完成事件的注册。

服务端完成bind操作后，使用运行方式启动客户端

直接快进到IO操作的处理

1.selector.select()方法停止阻塞，向下找到IO事件处理的方法

如果对存储SelectionKey的结构有优化，那么使用数组存储SelectionKey，则走第一个if分支

如果对存储SelectionKey的结构无优化，还是使用Set集合存储SelectionKey，则走第二个else分支

知识点回顾：

附件是什么？

附件就是NioChannel，可以是NioServerSocketChannel，也可以是NioSocketChannel。通过这些Channel才可以拿到对应的pipeline管道，通过管道我们才可以进行相应的读写IO，连接的网络操作。正是通过这种附件的方式，NIO才和Netty完美的整合在了一起。