Netty源码分析----NioEventLoop之任务队列

（*文章基于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实现的这个优先级队列我没看....不过理论上是这样，有兴趣的自己去看看吧=_=!)