死磕 java线程系列之线程池深刻解析——定时任务执行流程

时间 2019-11-09

标签 java 线程系列深刻解析定时任务执行流程栏目 Java 繁體版

原文原文链接

（手机横屏看源码更方便）java

注：java源码分析部分如无特殊说明均基于 java8 版本。面试

注：本文基于ScheduledThreadPoolExecutor定时线程池类。spring

简介

前面咱们一块儿学习了普通任务、将来任务的执行流程，今天咱们再来学习一种新的任务——定时任务。数据结构

定时任务是咱们常常会用到的一种任务，它表示在将来某个时刻执行，或者将来按照某种规则重复执行的任务。源码分析

问题

（1）如何保证任务是在将来某个时刻才被执行？学习

（2）如何保证任务按照某种规则重复执行？this

来个栗子

建立一个定时线程池，用它来跑四种不一样的定时任务。线程

public class ThreadPoolTest03 {
    public static void main(String[] args) throws ExecutionException, InterruptedException {
        // 建立一个定时线程池
        ScheduledThreadPoolExecutor scheduledThreadPoolExecutor = new ScheduledThreadPoolExecutor(5);

        System.out.println("start: " + System.currentTimeMillis());

        // 执行一个无返回值任务，5秒后执行，只执行一次
        scheduledThreadPoolExecutor.schedule(() -&gt; {
            System.out.println("spring: " + System.currentTimeMillis());
        }, 5, TimeUnit.SECONDS);

        // 执行一个有返回值任务，5秒后执行，只执行一次
        ScheduledFuture<string> future = scheduledThreadPoolExecutor.schedule(() -&gt; {
            System.out.println("inner summer: " + System.currentTimeMillis());
            return "outer summer: ";
        }, 5, TimeUnit.SECONDS);
        // 获取返回值
        System.out.println(future.get() + System.currentTimeMillis());

        // 按固定频率执行一个任务，每2秒执行一次，1秒后执行
        // 任务开始时的2秒后
        scheduledThreadPoolExecutor.scheduleAtFixedRate(() -&gt; {
            System.out.println("autumn: " + System.currentTimeMillis());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
        }, 1, 2, TimeUnit.SECONDS);

        // 按固定延时执行一个任务，每延时2秒执行一次，1秒执行
        // 任务结束时的2秒后，本文由公从号“彤哥读源码”原创
        scheduledThreadPoolExecutor.scheduleWithFixedDelay(() -&gt; {
            System.out.println("winter: " + System.currentTimeMillis());
            LockSupport.parkNanos(TimeUnit.SECONDS.toNanos(1));
        }, 1, 2, TimeUnit.SECONDS);
    }
}

定时任务整体分为四种：rest

（1）将来执行一次的任务，无返回值；code

（2）将来执行一次的任务，有返回值；

（3）将来按固定频率重复执行的任务；

（4）将来按固定延时重复执行的任务；

本文主要以第三种为例进行源码解析。

scheduleAtFixedRate()方法

提交一个按固定频率执行的任务。

public ScheduledFuture<!--?--> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    // 参数判断
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period &lt;= 0)
        throw new IllegalArgumentException();
        
    // 将普通任务装饰成ScheduledFutureTask
    ScheduledFutureTask<void> sft =
        new ScheduledFutureTask<void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    // 钩子方法，给子类用来替换装饰task，这里认为t==sft
    RunnableScheduledFuture<void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 延时执行
    delayedExecute(t);
    return t;
}

能够看到，这里的处理跟将来任务相似，都是装饰成另外一个任务，再拿去执行，不一样的是这里交给了delayedExecute()方法去执行，这个方法是干吗的呢？

delayedExecute()方法

延时执行。

private void delayedExecute(RunnableScheduledFuture<!--?--> task) {
    // 若是线程池关闭了，执行拒绝策略
    if (isShutdown())
        reject(task);
    else {
        // 先把任务扔到队列中去
        super.getQueue().add(task);
        // 再次检查线程池状态
        if (isShutdown() &amp;&amp;
            !canRunInCurrentRunState(task.isPeriodic()) &amp;&amp;
            remove(task))
            task.cancel(false);
        else
            // 保证有足够有线程执行任务
            ensurePrestart();
    }
}
void ensurePrestart() {
    int wc = workerCountOf(ctl.get());
    // 建立工做线程
    // 注意，这里没有传入firstTask参数，由于上面先把任务扔到队列中去了
    // 另外，没用上maxPoolSize参数，因此最大线程数量在定时线程池中实际是没有用的
    if (wc &lt; corePoolSize)
        addWorker(null, true);
    else if (wc == 0)
        addWorker(null, false);
}

到这里就结束了？！

实际上，这里只是控制任务能不能被执行，真正执行任务的地方在任务的run()方法中。

还记得上面的任务被装饰成了ScheduledFutureTask类的实例吗？因此，咱们只要看ScheduledFutureTask的run()方法就能够了。

ScheduledFutureTask类的run()方法

定时任务执行的地方。

public void run() {
    // 是否重复执行
    boolean periodic = isPeriodic();
    // 线程池状态判断
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 一次性任务，直接调用父类的run()方法，这个父类其实是FutureTask
    // 这里咱们再也不讲解，有兴趣的同窗看看上一章的内容
    else if (!periodic)
        ScheduledFutureTask.super.run();
    // 重复性任务，先调用父类的runAndReset()方法，这个父类也是FutureTask
    // 本文主要分析下面的部分
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 设置下次执行的时间
        setNextRunTime();
        // 重复执行，本文由公从号“彤哥读源码”原创
        reExecutePeriodic(outerTask);
    }
}

能够看到，对于重复性任务，先调用FutureTask的runAndReset()方法，再设置下次执行的时间，最后再调用reExecutePeriodic()方法。

FutureTask的runAndReset()方法与run()方法相似，只是其任务运行完毕后不会把状态修改成NORMAL，有兴趣的同窗点进源码看看。

再来看看reExecutePeriodic()方法。

void reExecutePeriodic(RunnableScheduledFuture<!--?--> task) {
    // 线程池状态检查
    if (canRunInCurrentRunState(true)) {
        // 再次把任务扔到任务队列中
        super.getQueue().add(task);
        // 再次检查线程池状态
        if (!canRunInCurrentRunState(true) &amp;&amp; remove(task))
            task.cancel(false);
        else
            // 保证工做线程足够
            ensurePrestart();
    }
}

到这里是否是豁然开朗了，原来定时线程池执行重复任务是在任务执行完毕后，又把任务扔回了任务队列中。

重复性的问题解决了，那么，它是怎么控制任务在某个时刻执行的呢？

OK，这就轮到咱们的延时队列登场了。

DelayedWorkQueue内部类

咱们知道，线程池执行任务时须要从任务队列中拿任务，而普通的任务队列，若是里面有任务就直接拿出来了，可是延时队列不同，它里面的任务，若是没有到时间也是拿不出来的，这也是前面分析中一上来就把任务扔进队列且建立Worker没有传入firstTask的缘由。

说了这么多，它究竟是怎么实现的呢？

其实，延时队列咱们在前面都详细分析过，想看完整源码分析的能够看看以前的《死磕 java集合之DelayQueue源码分析》。

延时队列内部是使用“堆”这种数据结构来实现的，有兴趣的同窗能够看看以前的《拜托，面试别再问我堆（排序）了！》。

咱们这里只拿一个take()方法出来分析。

public RunnableScheduledFuture<!--?--> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    // 加锁
    lock.lockInterruptibly();
    try {
        for (;;) {
            // 堆顶任务
            RunnableScheduledFuture<!--?--> first = queue[0];
            // 若是队列为空，则等待
            if (first == null)
                available.await();
            else {
                // 还有多久到时间
                long delay = first.getDelay(NANOSECONDS);
                // 若是小于等于0，说明这个任务到时间了，能够从队列中出队了
                if (delay &lt;= 0)
                    // 出队，而后堆化
                    return finishPoll(first);
                // 还没到时间
                first = null;
                // 若是前面有线程在等待，直接进入等待
                if (leader != null)
                    available.await();
                else {
                    // 当前线程做为leader
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 等待上面计算的延时时间，再自动唤醒
                        available.awaitNanos(delay);
                    } finally {
                        // 唤醒后再次得到锁后把leader再置空
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        if (leader == null &amp;&amp; queue[0] != null)
            // 至关于唤醒下一个等待的任务
            available.signal();
        // 解锁，本文由公从号“彤哥读源码”原创
        lock.unlock();
    }
}

大体的原理是，利用堆的特性获取最快到时间的任务，即堆顶的任务：

（1）若是堆顶的任务到时间了，就让它从队列中了队；

（2）若是堆顶的任务还没到时间，就看它还有多久到时间，利用条件锁等待这段时间，待时间到了后从新走（1）的判断；

这样就解决了能够在指定时间后执行任务。

其它

其实，ScheduledThreadPoolExecutor也是可使用execute()或者submit()提交任务的，只不过它们会被当成0延时的任务来执行一次。

public void execute(Runnable command) {
    schedule(command, 0, NANOSECONDS);
}
public <t> Future<t> submit(Callable<t> task) {
    return schedule(task, 0, NANOSECONDS);
}

总结

实现定时任务有两个问题要解决，分别是指定将来某个时刻执行任务、重复执行。

（1）指定某个时刻执行任务，是经过延时队列的特性来解决的；

（2）重复执行，是经过在任务执行后再次把任务加入到队列中来解决的。

彩蛋

到这里基本上普通的线程池的源码解析就结束了，这种线程池是比较经典的实现方式，总体上来讲，效率相对不是特别高，由于全部的工做线程共用同一个队列，每次从队列中取任务都要加锁解锁操做。

那么，能不能给每一个工做线程配备一个任务队列呢，在提交任务的时候就把任务分配给指定的工做线程，这样在取任务的时候就不须要频繁的加锁解锁了。

答案是确定的，下一章咱们一块儿来看看这种基于“工做窃取”理论的线程池——ForkJoinPool。

欢迎关注个人公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一块儿畅游源码的海洋。

</t></t></t></void></void></void></string>