什么是事件循环组
我们在Echo Server这个Netty的小Demo的启动代码中看到，无论是server端还是client端，一上来都实现创建单个事件循环组（客户端）或者父子事件循环组（服务端往往是创建两个，而且父循环组传入的参数是1，如果只绑定一个端口，那么填入1就好了），然后调用b.group()方法关联循环组。
那啥是事件循环组呢，从名字来推测，它的基本元素应该是EventLoop事件循环。可以说，NioEventLoop和NioEventLoopGroup是netty卓越的Reactor模型的重要体现。一个事件循环对应了一个Selector以及一个线程，用于接收处理IO事件。而事件循环组就是事件循环的管理者。在Server端，事件循环组往往是两个，一个主要用于连接的接收，而另外一个负责读写事件。
我们看一看NioEventLoopGroup的源码。跟踪NioEventLoopGroup的构造方法，会调用到下面的构造器，意思是当我们没有指定NioEventLoopGroup的构造器参数的时候，nThread=0，那么传入的线程数就变为cpu核心数乘以2.
protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
super(nThreads == 0? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args);
}
protected MultithreadEventExecutorGroup(int nThreads, ThreadFactory threadFactory, Object... args) {
if (nThreads <= 0) {
throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads));
}
if (threadFactory == null) {
threadFactory = newDefaultThreadFactory();
}
children = new SingleThreadEventExecutor[nThreads];
if (isPowerOfTwo(children.length)) {
chooser = new PowerOfTwoEventExecutorChooser();
} else {
chooser = new GenericEventExecutorChooser();
}
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(threadFactory, args);
success = true;
} catch (Exception e) {
// TODO: Think about if this is a good exception type
throw new IllegalStateException("failed to create a child event loop", e);
} finally {
// 创建失败时的处理
}
}
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
}
};
for (EventExecutor e: children) {
e.terminationFuture().addListener(terminationListener);
}
}
先调试看一下当程序执行到MultithreadEventExecutorGroup构造器时，传入的参数是啥。
然后看一下NioEventLoopGroup的构造方法中的逻辑：
1. 对线程数进行确定。如果没有指定线程数后者指定为0，则会将其修改为cpu核心数*2。
2.创建并初始化事件循环NioEventLoop >事件循环组中的每个元素为事件循环，在底层是通过一个名为children的数组（大小为传入的线程数）来保存事件循环的，这里children虽然是EventExecutor类型，但是我们实际使用NioEventLoopGroup的时候，newChild方法生成的children元素为NioEventLoop。
@Override
protected EventExecutor newChild(ThreadFactory threadFactory, Object... args) throws Exception {
return new NioEventLoop(this, threadFactory, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
跟踪NioEventLoop的构造方法（下方代码），发现一个NioEventLoop大概有下面这些参数：他所在的EventLoopGroup(parent字段)、对应的NIO中的Selector、选择策略、任务队列、拒绝策略、以及一个Thread实例。关于Executor和Thread是如何工作的，我们后边再去分析。这里稍微注意一下有两个Selector，一个是unwrapped，而另外一个selector（SelectedSelectionKeySetSelector）是对这个unwrapped的封装(装饰者模式)，实际它在工作的时候调用的还是内部封装的selector的方法。
NioEventLoop(NioEventLoopGroup parent, ThreadFactory threadFactory, SelectorProvider selectorProvider,
SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
super(parent, threadFactory, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler);
/*省略部分代码*/
provider = selectorProvider;
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
unwrappedSelector = selectorTuple.unwrappedSelector;
selectStrategy = strategy;
}
protected SingleThreadEventLoop(EventLoopGroup parent, ThreadFactory threadFactory,
boolean addTaskWakesUp, int maxPendingTasks,
RejectedExecutionHandler rejectedExecutionHandler) {
super(parent, threadFactory, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler);
}
protected SingleThreadEventExecutor(
EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp, int maxPendingTasks,
RejectedExecutionHandler rejectedHandler) {
this.parent = parent;
this.addTaskWakesUp = addTaskWakesUp;
thread = threadFactory.newThread(new Runnable() {
/*省略了run方法*/
});
threadProperties = new DefaultThreadProperties(thread);
this.maxPendingTasks = Math.max(16, maxPendingTasks);
taskQueue = newTaskQueue();
rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
}
3. 设置chooser。chooser实例具体为PowerOfTwoEventExecutorChooser或者GenericEventExecutorChooser，它的作用就在于选择一个children中的NioEventLoop，这两者的选择方式都是通过一个递增的计数值，选择出对应的children中的元素，具体都是 计数值 % 数组长度，但是当数组长度为2的次幂的时候，取余操作可以替换为 计数值 & (数组长度 - 1)。
4. 其他，设置了监听。
至此，NioEventLoopGroup以及其中的NioEventLoop就创建好了。
EventLoop中关联的thread及它是如何工作的
上文中我们注意到thread这个字段。SingleThreadEventExecutor构造器中指定了thread的run方法究竟要做什么。
thread = threadFactory.newThread(new Runnable() {
@Override
public void run() {
boolean success = false;
updateLastExecutionTime();
try {
// run方法由SingleThreadEventExecutor的具体子类所实现，
// 我们采用的是NioEventLoop，所以自然调用的就是NioEventLoop的run方法
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
/*省略部分代码*/
}
}
});
thread的run方法无非就是在干这个事：
SingleThreadEventExecutor.this.run();
而这个方法我们发现是需要被子类重写的，这里实际执行的就是NioEventLoop重写的run方法。关于run方法做了什么我们先不讲，后边会说。至此线程已经创建好了，但是发现构造器中并没有thread.start()让线程启动，那么线程是什么时候开始启动的呢？SingleThreadEventExecutor的execute(Runnable)方法会启动线程。
@Override
public void execute(Runnable task) {
if (task == null) {
throw new NullPointerException("task");
}
// 判断当前执行线程是否是NioEventLoop关联的线程。
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
// 启动关联的thread，调用它的start()方法
startThread();
// 将要执行的任务添加到任务队列
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
首先会判断当前执行线程是不是NioEventLoop关联的线程，是的话直接向任务队列中添加Runnable任务，不是的话（此时关联的线程还未启动），会将线程启动并将任务添加到任务队列。
我们现在已经比较清楚NioEventLoop中关联的线程的创建和启动时机，但是还没有细说它启动后要干嘛，所以我们转向NioEventLoop重写的run方法。
@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();
}
// fall through
default:
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
我现在发现NioEventLoop原来是一个死循环，也就是NioEventLoop中所关联的那个thread，一直就在执行run里面的循环。
我们从整体上来看一下这部分代码。hasTasks方法判断任务队列中是否含有任务（之前提到了当向NioEventLoop添加Runnable任务的时候，任务都会添加到任务队列），如果有任务，则执行selectNow()，没有任务的话执行select(wakenUp.getAndSet(false));(也就是进入case SelectStrategy.SELECT这个分支)
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
// wakeUp为true了表明需要唤醒阻塞在selector上的线程让他可以干别的事情
selector.wakeup();
}
// fall through
default:
}
当有任务队列有任务的时候，调用selectNow会触发NioEventLoop关联的Selector进行selectNow非阻塞调用（还记得上文说过的unwrappedSelector和它的包装实体selector吗，这里实际就是包装类最终调用的还是真实selector实例unwrappedSelector进行selectNow）。当没有任务的时候，NioEventLoop线程就可以多花点时间用于接收IO事件，可以稍微进行阻塞式的获取，但是又不能一直阻塞着，因为万一碰到任务队列中突然有了任务呢。所以，select(wakenUp.getAndSet(false))中在一定时间内阻塞select IO事件。稍微总结一下switch-case这部分，它也没干什么其他事儿，就是检测有没有IO事件或者有没有任务，这步执行完了之后，就要进行IO事件和任务的处理。
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
ioTime可以理解为分配给IO事件处理的时间占比，这段代码无非就是分别处理IO事件和任务。
runAllTasks和runAllTasks(long time)都用于执行任务，只不过后者有时间限制，它们做的就是将可以执行的Scheduled的任务(通过NioEventLoop的schedule方法可以添加定时任务)放入到taskQueue中，然后不断从taskQueue取任务执行。
processSelectedKeys()用于执行IO事件处理，有两个方法processSelectedKeysOptimized和processSelectedKeysPlain，这两个方法看起来不一样，但是内部原理差不多，我在调试的时候一直执行的都是processSelectedKeysOptimized，应该是selectedKeys所指代的对象跟selector实例关联起来了吧。
private void processSelectedKeysOptimized() {
for (int i = 0; i < selectedKeys.size; ++i) {
final SelectionKey k = selectedKeys.keys[i];
// null out entry in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
selectedKeys.keys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (needsToSelectAgain) {
// null out entries in the array to allow to have it GC'ed once the Channel close
// See https://github.com/netty/netty/issues/2363
selectedKeys.reset(i + 1);
selectAgain();
i = -1;
}
}
}
方法里面就是遍历每一个有效的SelectionKey，然后进行事件的处理，final Object a = k.attachment();是从SelectionKey实例中取出attachment，具体的attachment是什么是与netty的通道注册相关的。因为一个SelectionKey认为对应这一个NIO中的SocketChannel，那么这里的attachment取出的就是netty中封装了SocketChannel的那个NioSocketChannel实例（这一部分先了解一下，关于通道注册后面的文章会讲到）。正常情况下会进入processSelectedKey(k, (AbstractNioChannel) a)中执行，点进去会发现，这里面出现了很多Java NIO中比较熟悉的内容，然后在此基础上利用netty的组件（unsafe）进行其他的处理（这个后面会介绍）。
executor字段代表的Executor实例在执行execute(Runnable)方法时是直接调用ThreadPerTaskExecutor的execute(Runnable)来执行的，最终执行了传入的Runnable对象command的run方法完成具体执行逻辑。ThreadPerTaskExecutor执行execute(Runnable)方法时会创建一个新的线程来完成，所以实现了异步的过程。
@Override
public void execute(Runnable command) {
threadFactory.newThread(command).start();
}
我们大概已经了解了EventLoop中的executor是怎么进行工作的（它接收到Runnable任务之后最终创建一个新的线程来执行它），我们再来看看什么时候executor开始工作。找到NioEventLoop的execute方法（实际位于基类SingleThreadEventExecutor），主要代码如下
private void execute(Runnable task, boolean immediate) {
// inEventLoop 用于判断 当前实例的thread字段是否等于当前执行线程
// 第一次进入这个方法的时候由于thread还没有初始化，所以inEventLoop为false
boolean inEventLoop = inEventLoop();
// 向任务队列中插入要执行的Runnable任务
addTask(task);
if (!inEventLoop) {
// 如果没有设置当前EventLoop关联的线程thread，那么会进行线程的设置并启动相关逻辑执行
startThread();
if (isShutdown()) {
boolean reject = false;
try {
if (removeTask(task)) {
reject = true;
}
} catch (UnsupportedOperationException e) {
// The task queue does not support removal so the best thing we can do is to just move on and
// hope we will be able to pick-up the task before its completely terminated.
// In worst case we will log on termination.
}
if (reject) {
reject();
}
}
}
if (!addTaskWakesUp && immediate) {
wakeup(inEventLoop);
}
}
通过以上的代码我们可以知道，当我们向一个NioEventLoop塞入一个Runnable任务的时候，它会把这个执行任务添加到任务队列TaskQueue中非阻塞的返回，如果当前NioEventLoop还没有绑定线程的时候，会设置线程并启动线程，也就是startThread方法。也就是说当NioEventLoop已经关联到一个线程的时候，每次向它提交一个任务只是简单地非阻塞式的将任务添加到NioEventLoop的任务队列中，方法就返回了。
我们重点看一下线程启动的时候都在做什么，也就是startThread()。重点在doStartThread中。它会调用所关联的executor来执行一个Runnable任务，由上文可知，executor实例会借助ThreadPerTaskExecutor创建一个新的线程来执行这个Runnable任务，所以thread = Thread.currentThread();这条语句表明当前NioEventLoop关联的Thread实例就是新创建的线程。而Runnable任务所执行的就是被SingleThreadEventExecutor子类(这里就是NioEventLoop)重写的run()方法。
private void doStartThread() {
assert thread == null;
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
// 这里的run()方法被SingleThreadEventExecutor子类实例重写，
// 例如我们采用的是NioEventLoop，那么这里执行的就是NioEventLoop重写的run方法
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {
logger.warn("Unexpected exception from an event executor: ", t);
} finally {
/*线程结束run方法后的终止逻辑，暂时忽略*/
}
}
});
}
*链接
1. Netty解析：第一个demo——Echo Server
2. Netty解析：NioEventLoopGroup事件循环组
3. Netty解析：NioSocketChannel、NioServerSocketChannel的创建及注册
4. Netty解析：Handler、Pipeline大动脉及其在注册过程中体现
5. Netty解析：connect/bind方法背后
6. Netty解析：服务端如何接受连接并后续处理读写事件
