netty源码阅读之NioEventLoop之NioEventLoop执行-----runAllTask

太过爱你忘了你带给我的痛 2022-05-17 10:41 154阅读 0赞

从《[netty源码阅读之NioEventLoop之NioEventLoop执行][netty_NioEventLoop_NioEventLoop]》知道，select之后，有两个步骤：

processSelectedKey()和runAllTask()

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);
                }

这里有个ioRatio，就是执行io任务和非io任务的时间比。用户可以自行设置。默认为50，可以看源码。如果是100的话，先执行完io任务，执行完之后才执行非io任务。我们现在学习比较复杂的第二种情况。

在runAllTasks(long timeoutNanos)里面，主要分享以下三件事情：

1、task的分类和添加

2、任务的聚合（把定时任务队列里面任务聚合到mpscQueue里面）

3、任务的真正执行

点进去runAllTasks(long timeoutNanos)方法的实现：

protected boolean runAllTasks(long timeoutNanos) {
            fetchFromScheduledTaskQueue();
            Runnable task = pollTask();
            if (task == null) {
                afterRunningAllTasks();
                return false;
            }
    
            final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
            long runTasks = 0;
            long lastExecutionTime;
            for (;;) {
                safeExecute(task);
    
                runTasks ++;
    
                // Check timeout every 64 tasks because nanoTime() is relatively expensive.
                // XXX: Hard-coded value - will make it configurable if it is really a problem.
                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;
        }

netty里面默认有两个任务队列，一个定时任务队列，一个我们之前说过的mpscQueue。

第一步是把定时任务读出来， 第二步是从任务队列里面取任务，第三步是safeExecute()执行任务，通过(runTasks & 0x3F) == 0判断执行的任务次数是否达到64个，如果到达，那么就判断是否到达我们非io任务的时间，是的话就退出，并且记录下最后执行任务的时间lastExecutionTime，否则继续取任务去执行。所有任务执行完了，会调用afterRunningAllTasks()进行收尾工作。

关于64，在源码里面也就解释，为什么不执行完一次任务检查一次？检查（nanoTime的执行）也是相对耗时的操作，这里他们用了硬编码的方式，如果实在需要，可以修改为可配置的。

我们首先分析fetchFromScheduledTaskQueue()

private boolean fetchFromScheduledTaskQueue() {
            long nanoTime = AbstractScheduledEventExecutor.nanoTime();
            Runnable scheduledTask  = pollScheduledTask(nanoTime);
            while (scheduledTask != null) {
                if (!taskQueue.offer(scheduledTask)) {
                    // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
                    scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
                    return false;
                }
                scheduledTask  = pollScheduledTask(nanoTime);
            }
            return true;
        }

也就是每次从定时任务队列scheduledTaskQueue里面取出到期的任务，添加到我们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;
        }

在这里可以看到，到期的定时任务会返回并从定时任务队列里面删除。最快到期的定时任务会在最前面，关于这个定时任务队列，我们继续看this.scheduledTaskQueue这个的实现：

Queue<ScheduledFutureTask<?>> scheduledTaskQueue() {
            if (scheduledTaskQueue == null) {
                scheduledTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();
            }
            return scheduledTaskQueue;
        }

使用了优先队列的实现，所以可以比较。而比较的方法在它的定时任务ScheduledFutureTask里面：

@SuppressWarnings("ComparableImplementedButEqualsNotOverridden")
    final class ScheduledFutureTask<V> extends PromiseTask<V> implements ScheduledFuture<V> {
        ...
    
        @Override
        public int compareTo(Delayed o) {
            if (this == o) {
                return 0;
            }
    
            ScheduledFutureTask<?> that = (ScheduledFutureTask<?>) o;
            long d = deadlineNanos() - that.deadlineNanos();
            if (d < 0) {
                return -1;
            } else if (d > 0) {
                return 1;
            } else if (id < that.id) {
                return -1;
            } else if (id == that.id) {
                throw new Error();
            } else {
                return 1;
            }
        }
    
        ...
    }

可以看到，deadline时间最小的，是排在最前面的，如果deadline时间相同，那么就比较id，可以严格保证任务的先后顺序。

关于NioEventLoop的定时任务的实现，还有一个细节，在它父类AbstractScheduledEventExecutor的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;
        }

如果丢定时任务的线程是NioEventLoop的线程，那么就把它放到定时任务队列里面，这是添加定时任务，但是我们这里关注的是，如果不是NioEventLoop的线程，那就会调用execute方法，新建一个线程来把任务丢到定时任务队列，这个新建的线程最终会绑定到NioEventLoop，通过这种方式保证了线程的安全。细细体会作者这种方法，特别巧妙。

接下去就是Runnable task = pollTask();，进入这个方法，最终可以看到就是从我们的taskQueue里面把在最前面的任务取出来，这里的任务已经包括可能的定时任务了。

protected final Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
            for (;;) {
                Runnable task = taskQueue.poll();
                if (task == WAKEUP_TASK) {
                    continue;
                }
                return task;
            }
        }

最后是safeExecute(task)，直接调用task的run方法（从这里可以知道，其实线程池或者线程执行的runable的时候，其实就是执行runable的run方法而已！）：

/**
         * Try to execute the given {@link Runnable} and just log if it throws a {@link Throwable}.
         */
        protected static void safeExecute(Runnable task) {
            try {
                task.run();
            } catch (Throwable t) {
                logger.warn("A task raised an exception. Task: {}", task, t);
            }
        }

如果执行错误，会打印日志，不影响别的任务的使用。有的源码的实现，一个任务不行，别的任务也都崩溃了。

[netty_NioEventLoop_NioEventLoop]: https://blog.csdn.net/fst438060684/article/details/81558585