之前在bootStrap去注册通道的时候，顺便粗略的过了一下NioEventLoop的工作过程。只能大概看清楚工作流程，但是其原理，还是没有很细致的说到，所以这次还是详细看看，NioEventLoop是如何工作的，其实本质上，只是看Netty如何去维护这些channel的读写以及注册操作。
入口还是之前的demo，代码如下
public class TimeClient {
public static void main(String args[]) {
connect();
}
private static void connect() {
//用于客户端处通道的读写
EventLoopGroup work = new NioEventLoopGroup();
Bootstrap b = new Bootstrap();
b.group(work).option(ChannelOption.TCP_NODELAY, true).channel(NioSocketChannel.class)
.handler(new TimeClientHandler());
ChannelFuture cf = null;
try {
//一直阻塞，直到连接上服务端
cf = b.connect(ConnectConfig.getHost(), ConnectConfig.getPort()).sync();
//一直阻塞，直到该通道关闭
cf.channel().closeFuture().sync();
} catch (InterruptedException e) {
e.printStackTrace();
} finally {
//避免线程没有杀死
work.shutdownGracefully();
}
}
}
NioEventLoopGroup
为什么说这个类呢？可以先解释一下，为什么说这个类，因为在实际的代码运行中，netty是通过这个类，去维护channel的读写以及注册功能的。然后在此之前，要先讲一下NioEventLoopGroup这个类，NioEventLoopGroup.这个链接中有基本的结构，就不过多说了。
基本的接口
1.EventExecutorGroup
其实从接口结构上看是比较简单的
具备的功能(管理EventExecutor)
1.说明具备了线程池的基本功能，继承了ScheduledExecutorService接口
2.新加了几个方法，主要是用于管理EventExecutor的。
boolean isShuttingDown();
Future<?> shutdownGracefully();
Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit);
Future<?> terminationFuture();
EventExecutor next();
2.EventExecutor
其实从接口结构上看是比较简单的, 为EventExecutorGroup的子类
具备的功能，因为有一些地方还木有理解，就说比较关键的点。
1.获取工作组，具体体现在以下方法中
EventExecutorGroup parent()；
2.判断某个线程是否与EventExecutor处于同一个线程。
boolean inEventLoop();
3.EventLoopGroup
为EventExecutor的子类，另外新增了几个方法。
新增的功能：
1.将获取的类型改为EventLoop(为EventLoopGroup的子类)
@Override
EventLoop next();
2.注册通道
ChannelFuture register(Channel channel);
4.EventLoop
为EventLoopGroup的子类，没有太特殊的地方
有以下的方法:
@Override
EventLoopGroup parent();
结构上来看：EventExecutorGroup -》EventExecutor -》EventLoopGroup -》EventLoop
最后可以看出EventLoop大概是这么一个东西，具备判断线程池的功能，注册通道，还有获取自身对应的parent（Group）,以及当前线程与EventLoop所绑定的线程是否一致(这里其实是为了减少资源竞争，一个channel的相关维护，只会被一个线程所处理)。
EventLoop是如何维护channel的读写以及注册的？
NioEventLoop的结构，以及生成
首先要从NioEventLoopGroup的构造方法入手
public class NioEventLoopGroup extends MultithreadEventLoopGroup {
public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
final SelectStrategyFactory selectStrategyFactory) {
传入线程数，SelectorProvider，还有一个选择策略工厂，以及拒绝策略(当任务队列过长的时候)
super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
}
}
public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup {
private static final int DEFAULT_EVENT_LOOP_THREADS;
static {
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
}
默认的eventLoop线程数，为核心线程数*2
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
}
//到最后定位到此处。。。将无关代码省略了。为了方便阅读
public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
//该group管理的EventExecutor
private final EventExecutor[] children;
//EventExecutor的选择器
private final EventExecutorChooserFactory.EventExecutorChooser chooser;
//暴露给子类，用于实例化EventExecutor
protected abstract EventExecutor newChild(Executor executor, Object... args) throws Exception;
protected MultithreadEventExecutorGroup(int nThreads, Executor executor,
EventExecutorChooserFactory chooserFactory, Object... args) {
//executor的类型为ThreadPerTaskExecutor，其实就是该executor每执行一个任务，都会创建一个线程去执行
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
//初始化数组
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
关键点在这里，这里涉及到EventExecutor的实例化。
newChild是一个抽象方法，在子类中被实现。
那么看看是如何实例化的，具体实现位于子类NioEventLoop中..........................................................................................................................................
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {
throw new IllegalStateException("failed to create a child event loop", e);
}
设置EventLoop的选择器，用于获取EventLoop。
chooser = chooserFactory.newChooser(children);
}
}
EventExecutor的实例化，newChild方法的实现, 在EventLoopGroup中被覆写
public class NioEventLoopGroup extends MultithreadEventLoopGroup {
@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
return new NioEventLoop(this, executor, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
}
public final class NioEventLoop extends SingleThreadEventLoop {
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider,
SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
设置拒绝策略，以及任务队列的长度，以及线程池的类型，还是直接往里面看..............
父类SingleThreadEventExecutor的构造方法
super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
if (selectorProvider == null) {
throw new NullPointerException("selectorProvider");
}
if (strategy == null) {
throw new NullPointerException("selectStrategy");
}
//这个provider是单例，所以可以确保这个工作组的selector是公用同一个
provider = selectorProvider;
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
unwrappedSelector = selectorTuple.unwrappedSelector;
selectStrategy = strategy;
}
}
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
boolean addTaskWakesUp, int maxPendingTasks,
RejectedExecutionHandler rejectedHandler) {
super(parent);
this.addTaskWakesUp = addTaskWakesUp;
this.maxPendingTasks = Math.max(16, maxPendingTasks);
//执行器，其实说白就是通过executor来创建线程。netty没那么傻，肯定是有变量去判断这个eventLoop是否已经持有线程了。
//因为executor的类型是ThreadPerTaskExecutor。感兴趣看看内部实现
this.executor = ObjectUtil.checkNotNull(executor, "executor");
//设置任务队列
taskQueue = newTaskQueue(this.maxPendingTasks);
//拒绝策略
rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
}
}
ThreadPerTaskExecutor的内部实现，每执行一个任务创建一个线程。
public final class ThreadPerTaskExecutor implements Executor {
private final ThreadFactory threadFactory;
public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
if (threadFactory == null) {
throw new NullPointerException("threadFactory");
}
this.threadFactory = threadFactory;
}
@Override
public void execute(Runnable command) {
threadFactory.newThread(command).start();
}
}
从上面的代码来看。可以得知，NioEventLoop具体有的功能如下：
1.创建线程(说白了就是每个NioEventLoop初始化的时候都是没对应的线程的，所以需要创建)
2.有任务队列(NioEventLoop去执行任务的时候，可以用于临时存储任务)
3.有拒绝策略(任务队列满了就拒绝)
NioEventLoop是如何去执行任务的
接下来，举个例子，看看是如何执行任务的。刚刚我们说到NioEventLoop的基本功能。
那么就从channel的注册，到连接，看看如何保证一个channel，只会被一个NioEventLoop处理。
下面还是先从注册开始吧
注册
注册通道的入口
public class Bootstrap extends AbstractBootstrap<Bootstrap, Channel> {
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
注册是在这里执行的======================注册是使用NioEventLoop去注册的，所以重点看此方法。
最后实现其实是在SingleThreadEventLoop，单线程的eventLoop，看看下面的代码片段。
拿到注册的回调，因为注册是异步运行的。
所以注册的回调有几种情况，注册完成了，注册没完成。
针对注册没完成，还有一些操作，所以加了监听器，在注册完成的时候会去调用监听器的方法。
而注册成功的，直接去连接就好了。
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
若注册成功
if (regFuture.isDone()) {
if (!regFuture.isSuccess()) {
return regFuture;
}
连接服务端
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else {
若注册没成功，先添加监听器，等到注册完成的时候，再去进行连接服务端
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
promise.setFailure(cause);
} else {
promise.registered();
doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
}
}
});
return promise;
}
}
}
如何注册通道
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
@Override
public ChannelFuture register(final ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
通过unsafe来注册，Unsafe是一个内部类，这里不关心它是啥，只看是怎么注册的。
通过debug，看到是在内部类中做的，直接看下面的代码片段。
promise.channel().unsafe().register(this, promise);
return promise;
}
}
注册的实现。
public abstract class AbstractChannel extends DefaultAttributeMap implements Channel {
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
//eventLoop为空，则抛异常
if (eventLoop == null) {
throw new NullPointerException("eventLoop");
}
//通道注册了没？不得二次注册，这里netty的channel有一个标记用于判断channel是否已经注册
if (isRegistered()) {
promise.setFailure(new IllegalStateException("registered to an event loop already"));
return;
}
//这里没细看 估计是eventLoop的类型与通道是不是匹配，不是那么重要，跳过。。。。
if (!isCompatible(eventLoop)) {
promise.setFailure(
new IllegalStateException("incompatible event loop type: " + eventLoop.getClass().getName()));
return;
}
//给这个channel设置对应的eventLoop，说明这个channel被这个eventLoop
AbstractChannel.this.eventLoop = eventLoop;
重点，判断当前线程是否与eventLoop中的线程是一个线程，是的话就同步执行好了。一开始当然不是
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
通过eventLoop的执行方法去执行注册。
同步的就没必要看了，因为我们不关心它怎么注册，而是线程如何去执行这个注册的task
======================================直接debug，定位到位置
SingleThreadEventExecutor的execute方法。直接看下一个代码片段。
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {
closeForcibly();
closeFuture.setClosed();
//给回调设置异常信息
safeSetFailure(promise, t);
}
}
}
}
SingleThreadEventExecutor的execute方法
其实下面这个方法，看上去已经简单明了了。
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements
OrderedEventExecutor {
默认eventLoop没分配线程，所以是not_started状态，避免分配多个线程。
private volatile int state = ST_NOT_STARTED;
eventLoop对应的所分配的线程变量
private volatile Thread thread;
12345 分别对应几个状态
private static final int ST_NOT_STARTED = 1;
private static final int ST_STARTED = 2;
private static final int ST_SHUTTING_DOWN = 3;
private static final int ST_SHUTDOWN = 4;
private static final int ST_TERMINATED = 5;
任务队列
private final Queue<Runnable> taskQueue;
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
判断当前线程，与eventLoop分配的线程是不是同一个。
boolean inEventLoop = inEventLoop();
如果是同一个线程，则直接加任务即可。一开始eventLoop没分配线程，所以必然需要开启线程。
if (inEventLoop) {
addTask(task);
} else {
开启线程=============================这是很重要的方法
下面的方法会提到
startThread();
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
private void startThread() {
//比较简单的写法，双重锁
if (state == ST_NOT_STARTED) {
//cas 无锁
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
这是重点了===============================================
doStartThread();
}
}
}
private void doStartThread() {
executor.execute(new Runnable() {
@Override
public void run() {
boolean success = false;
updateLastExecutionTime();
try {
这里我们可以猜一下，由于线程起来了，要一直运行，所以里面必然是一个死循环。
只有EventLoop状态为关闭的时候，才会跳出这个循环。
所以接下来直接看看这个run方法。
============================================================
这个run方法子类已经实现了，在NioEventLoop中，所以直接定位到这个位置。看下面的代码片段。
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
下面省略了一部分，其实是EventLoop关闭的时候，怎么去修改状态的内容。相对来说比较简单，所以直接忽略了。
}
}
});
}
}
线程是如何去维护channel的读写以及各种任务的
NioEventLoop中的run方法就实现了。
在注册的时候，channel会注册到eventLoop中的selector中。所以eventloop只需要通过分配到的线程，去做一个死循环，轮询selector中就绪的事件即可。这样子就可以维护多个channel了。一个channel只能被一个eventLoop维护，而一个eventLoop可以维护多个channel。
public final class NioEventLoop extends SingleThreadEventLoop {
@Override
protected void run() {
for (;;) {
try {
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
处理就绪的事件，被该eventLoop所维护的channel。因为注册的时候会将相关的selectkey都存到对应的selectkey集合中，位于父类的成员变量中。
processSelectedKeys();
} finally {
执行队列里面的任务
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
处理就绪的事件
processSelectedKeys();
} finally {
final long ioTime = System.nanoTime() - ioStartTime;
执行队列里面的任务，其实就是成员变量里面的taskQueue，在父类中SingleThreadEventExecutor
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
对EventLoop关闭的处理，如果正则关闭，则该死循环结束。
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
}
注册完成后，如何连接呢？
先回到注册通道的入口
public class Bootstrap extends AbstractBootstrap<Bootstrap, Channel> {
private ChannelFuture doResolveAndConnect(final SocketAddress remoteAddress, final SocketAddress localAddress) {
注册是在这里执行的======================注册是使用NioEventLoop去注册的，所以重点看此方法。
最后实现其实是在SingleThreadEventLoop，单线程的eventLoop，看看下面的代码片段。
拿到注册的回调，因为注册是异步运行的。
所以注册的回调有几种情况，注册完成了，注册没完成。
针对注册没完成，还有一些操作，所以加了监听器，在注册完成的时候会去调用监听器的方法。
而注册成功的，直接去连接就好了。
final ChannelFuture regFuture = initAndRegister();
final Channel channel = regFuture.channel();
若注册成功
if (regFuture.isDone()) {
if (!regFuture.isSuccess()) {
return regFuture;
}
连接服务端,看看服务端的代码
return doResolveAndConnect0(channel, remoteAddress, localAddress, channel.newPromise());
} else {
若注册没成功，先添加监听器，等到注册完成的时候，再去进行连接服务端
final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
Throwable cause = future.cause();
if (cause != null) {
promise.setFailure(cause);
} else {
promise.registered();
doResolveAndConnect0(channel, remoteAddress, localAddress, promise);
}
}
});
return promise;
}
}
如何连接？最后注册完，一路debug，到这里。
从这里可以看到，最后还是通过通道对应的eventLoop来。所以从通道的注册，到连接，以及处理通道中就绪的事件，一切都是在某个eventLoop中来维护的。
这种做法就避免了一个channel的事件被多个线程执行，不存在资源竞争的关系。
private static void doConnect(
final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise connectPromise) {
final Channel channel = connectPromise.channel();
channel.eventLoop().execute(new Runnable() {
@Override
public void run() {
if (localAddress == null) {
channel.connect(remoteAddress, connectPromise);
} else {
channel.connect(remoteAddress, localAddress, connectPromise);
}
connectPromise.addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
}
});
}
}
概括一下：
注册操作：
channel的注册，会被某一个eventLoop所执行，同时注册到该eventLoop的selector中。同时channel会设置自己的eventLoop，即该channel只会被该eventLoop所维护。
连接操作：
channel获取自己的eventLoop，然后将连接任务丢到任务队列中，由eventLoop分配的线程去处理。
读写操作：
eventLoop分配的线程，轮询selector中的key，进行处理。
