（*文章基于Netty4.1.22版本）
这篇文章主要分析一下NioEventLoop中任务队列相关的代码
源码分析
taskQueue
execute方法调用的时候或者执行完startThread方法，会调用addTask添加任务
protected void addTask(Runnable task) {
if (!offerTask(task)) {
reject(task);
}
}
final boolean offerTask(Runnable task) {
if (isShutdown()) {
reject();
}
return taskQueue.offer(task);
}
可以看到execute方法只是加到队列，异步的去执行这个任务，Netty的线程执行任务的时候，有个规律：
- 一部分时间执行IO任务，一部分时间执行非IO任务
队列中的任务，属于非IO任务，所以添加之后也不会是马上执行，等待非IO时间到了才会去执行(后续具体分析)
taskQueue任务的执行
主要是runAllTasks方法
protected boolean runAllTasks() {
assert inEventLoop();
boolean fetchedAll;
boolean ranAtLeastOne = false;
do {
//将到期的延时任务放到队列中
fetchedAll = fetchFromScheduledTaskQueue();
if (runAllTasksFrom(taskQueue)) {// 从taskQueue中获取任务并执行
ranAtLeastOne = true;
}
} while (!fetchedAll); // keep on processing until we fetched all scheduled tasks.
if (ranAtLeastOne) {//更新lastExecutionTime
lastExecutionTime = ScheduledFutureTask.nanoTime();
}
// 执行完taskQueue中的任务后的一些操作
afterRunningAllTasks();
return ranAtLeastOne;
}
看下runAllTasksFrom是如何处理的任务的
protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
Runnable task = pollTaskFrom(taskQueue);//taskQueue.poll()
if (task == null) {// 任务已经没了
return false;
}
for (;;) {// 不停的从taskQueue中获取任务并调用任务的run方法
safeExecute(task);// task.run
task = pollTaskFrom(taskQueue);
if (task == null) {
return true;
}
}
}
还有一个是SingleThreadEventExecutor.runAllTasks(long)方法
protected boolean runAllTasks(long timeoutNanos) {
fetchFromScheduledTaskQueue();// 同上
Runnable task = pollTask();
if (task == null) {
afterRunningAllTasks();
return false;
}
// 这个runTask方法和上面不一样，不是无限制的执行
// 而是有一定的时间限制，deadLine就是时间限制
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;// 执行的任务数
long lastExecutionTime;
for (;;) {
safeExecute(task);// 执行任务
runTasks ++;
//每执行16个任务，判断一下时间是否到达限制，到达了限制之后就不再执行任务
if ((runTasks & 0x3F) == 0) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
task = pollTask();
if (task == null) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
break;
}
}
afterRunningAllTasks();
this.lastExecutionTime = lastExecutionTime;
return true;
}
两个runAllTask方法区别就在于后者有一定的时间片限制，就如之前说的Netty一部分时间执行IO任务，一部分时间执行非IO任务，这个和某个变量相关，根据不同的值执行不同的方法
tailTasks
这个也是任务队列，这个任务通过SingleThreadEventLoop.executeAfterEventLoopIteration方法添加一个任务(但是目前没看到使用的地方，应该是Netty开放的扩展点)，和taskQueue不同的是，他的优先级没有taskQueue高，看下执行任务的代码，runAllTasks中当taskQueue的任务执行完毕或者时间限制到达后会调用afterRunningAllTasks方法，这个方法如下
//SingleThreadEventLoop.afterRunningAllTasks()
protected void afterRunningAllTasks() {
runAllTasksFrom(tailTasks);
}
即当taskQueue的任务执行完毕后才会去执行tailTasks中的任务
scheduledTaskQueue
scheduledTaskQueue从名字上看，就知道他存放的是延时任务，那么工作机制是如何呢？
先看下如何使用，如下代码执行一个延时任务
NioEventLoopGroup group = new NioEventLoopGroup(3);
group.next().schedule(()->{
System.out.println("run....");
}, 5, TimeUnit.SECONDS);
非常简单，调用EventLoop的schedule方法就OK了，和java的Timer类似。
看下schedule方法如何执行
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
//....
return schedule(new ScheduledFutureTask<Void>(
this, command, null, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
}
会将Runnable对象封装成Netty内部的一个延时任务对象，包含了该Runnable和任务的deadline，看下schedule方法
<V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
if (inEventLoop()) {
scheduledTaskQueue().add(task);
} else {
execute(new Runnable() {
@Override
public void run() {
scheduledTaskQueue().add(task);
}
});
}
return task;
}
有两个分支流程，最后都是将任务放到scheduledTaskQueue中，先看下scheduledTaskQueue是什么类型的队列
PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue() {
if (scheduledTaskQueue == null) {
scheduledTaskQueue = new DefaultPriorityQueue<ScheduledFutureTask<?>>(
SCHEDULED_FUTURE_TASK_COMPARATOR,
// Use same initial capacity as java.util.PriorityQueue
11);
}
return scheduledTaskQueue;
}
DefaultPriorityQueue是Netty的一个优先级队列，他是一个非线程安全的队列，那么在多线程情况下，不就有问题咯？
这时候看下上面的两个分支条件：
inEventLoop为false和inEventLoop为true，inEventLoop方法说过很多次，如果当前线程是EventLoop线程，inEventLoop则返回true，如果当前线程是非EventLoop线程，那么就返回false。
针对这两种情况，Netty做了不同处理：
- inEventLoop为true：直接调用scheduledTaskQueue队列的add方法
- inEventLoop为false：通过execute方法，添加一个任务到taskQueue队列中，
这个任务的作用就是调用scheduledTaskQueue队列的add方法
为什么在非EventLoop线程需要通过任务去添加到队列？我们知道taskQueue队列是一个线程安全的队列(他的实现是MPSC队列，具体原理没研究过，有空学习一下再分析，这里只需只有这个队列时线程安全的)，而taskQueue队列又只是只有EventLoop线程内部执行，所以这个任务是线程安全的。
就这样Netty非常巧妙的将一个非线程安全的队列的操作，转换成任务放到一个线程安全的队列中，让EventLoop线程去执行，而EventLoop只有一个线程，add的过程也不存在并发问题
延时任务的执行
接下来要看下延时的任务如何去执行，回到刚刚的地方，将任务加到队列后，貌似就没有其他的执行操作了，但是回顾一下taskQueue队列的执行过程，在执行前就是先去取延时任务，所以看下taskQueue队列任务执行的地方，直接看调用的fetchFromScheduledTaskQueue方法
private boolean fetchFromScheduledTaskQueue() {
long nanoTime = AbstractScheduledEventExecutor.nanoTime();
Runnable scheduledTask = pollScheduledTask(nanoTime);
while (scheduledTask != null) {
if (!taskQueue.offer(scheduledTask)) {
// 如果taskQueue满了放不进去，那么再重新放回延时队列中等待重新执行.
scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
return false;
}
scheduledTask = pollScheduledTask(nanoTime);
}
return true;
}
延时任务的一般原理判断队列中的任务的deadline是否大于等于当前时间，所以理所当然第一步就需要获取当前时间，并通过pollScheduledTask去取任务，当取到了任务，并不是直接执行，而是把该任务放到了taskQueue中(fetchFromScheduledTaskQueue方法后面的步骤就是从taskQueue中取任务执行)。
- 为什么不是直接执行呢？
我的理解是这样的：
以前说过NioEventLoop线程中分两部分时间，一部分时间是处理IO时间，一部分时间执行非IO任务，在前面我们可以看到从taskQueue取出任务之后会计算时间是否达到时间限制，那么如果这里直接执行，我觉得当然也可以，但是计算就会比较复杂，需要将两部分逻辑的时间相加，现在的处理方式就非常的通用了，只需要计算taskQueue的执行时间。还有一点就是直接执行的话，可能延时任务会阻碍普通任务的执行
接下来看下pollScheduledTask方法
protected final Runnable pollScheduledTask(long nanoTime) {
assert inEventLoop();
Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : scheduledTaskQueue.peek();
if (scheduledTask == null) {
return null;
}
if (scheduledTask.deadlineNanos() <= nanoTime) {
scheduledTaskQueue.remove();
return scheduledTask;
}
return null;
}
恩，和上面说的延时队列的原理一样，从队列中获取任务判断是否到达执行时间。
注意，由于scheduledTaskQueue是优先级队列，那么peek出来的就是时间比较前面的(为什么用应该，因为Netty实现的这个优先级队列我没看....不过理论上是这样，有兴趣的自己去看看吧=_=!)
网友评论