定时任务之ScheduledThreadPoolExecutor

java

java自带的定时任务执行器为ScheduledThreadPoolExecutor，继承于线程池管理器ThreadPoolExecutor，常用方法如下，程序将每2秒输出一次系统时间。

1
2
3
4
5
6
7
8
9
ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(2);
executor.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        System.out.println(DateFormatUtils.format(System.currentTimeMillis(), "HH:mm:ss SSS"));
    }
}, 0, 2, TimeUnit.SECONDS);
Thread.sleep(10_000);
executor.shutdown();

java.util.concurrent.ScheduledThreadPoolExecutor#scheduleAtFixedRate方法把参数Runnable对象封装成自定义的定时任务java.util.concurrent.ScheduledThreadPoolExecutor.ScheduledFutureTask。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
public ScheduledFuture<?> scheduleAtFixedRate(Runnable command,
                                              long initialDelay,
                                              long period,
                                              TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    if (period <= 0)
        throw new IllegalArgumentException();
    // 包装Runnable对象，用于有序阻塞列队DelayedWorkQueue的排序
    ScheduledFutureTask<Void> sft =
        new ScheduledFutureTask<Void>(command,
                                      null,
                                      triggerTime(initialDelay, unit),
                                      unit.toNanos(period));
    RunnableScheduledFuture<Void> t = decorateTask(command, sft);
    sft.outerTask = t;
    // 把任务放入执行队列，并检查ScheduledThreadPoolExecutor线程数量
    // 如果当前线程池线程数量小于线程池定义的核销线程数量，创建Worker添加新线程
    delayedExecute(t);
    return t;
}

ScheduledThreadPoolExecutor的阻塞队列为其自实现的DelayedWorkQueue，实际上是一个按定时任务RunnableScheduledFuture的下次执行时间排序的有序队列。

在ScheduledThreadPoolExecutor中每次添加一个定时任务时，调用方法java.util.concurrent.ScheduledThreadPoolExecutor.DelayedWorkQueue#offer(java.lang.Runnable)向DelayedWorkQueue添加元素。DelayedWorkQueue的具体实现如下，如果队列为空，则元素直接置入为队列首位；如果队列不为空，则队列以任务的下一次执行时间顺序排序。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
public boolean offer(Runnable x) {
    if (x == null)
        throw new NullPointerException();
    // DelayedWorkQueue的入参必须为定时任务RunnableScheduledFuture
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        int i = size;
        // 队列扩容
        if (i >= queue.length)
            grow();
        size = i + 1;
        if (i == 0) {
            // 空队列，直接为队列首位
            queue[0] = e;
            setIndex(e, 0);
        } else {
            // 按定时任务下次执行时间顺序排序
            siftUp(i, e);
        }
        if (queue[0] == e) {
            leader = null;
            available.signal();
        }
    } finally {
        lock.unlock();
    }
    return true;
}

private void siftUp(int k, RunnableScheduledFuture<?> key) {
    while (k > 0) {
        int parent = (k - 1) >>> 1;
        RunnableScheduledFuture<?> e = queue[parent];
        if (key.compareTo(e) >= 0)
            break;
        queue[k] = e;
        setIndex(e, k);
        k = parent;
    }
    queue[k] = key;
    setIndex(key, k);
}

ScheduledThreadPoolExecutor添加Worker创建线程调用的父类的方法java.util.concurrent.ThreadPoolExecutor#addWorker，然后新建的线程程调用方法java.util.concurrent.ThreadPoolExecutor#runWorker读取阻塞队列获取待执行的任务。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
final void runWorker(Worker w) {
    Thread wt = Thread.currentThread();
    Runnable task = w.firstTask;
    w.firstTask = null;
    w.unlock(); // allow interrupts
    boolean completedAbruptly = true;
    try {
        // 核心线程循环阻塞-执行任务
        while (task != null || (task = getTask()) != null) {
            w.lock();
            // If pool is stopping, ensure thread is interrupted;
            // if not, ensure thread is not interrupted.  This
            // requires a recheck in second case to deal with
            // shutdownNow race while clearing interrupt
            if ((runStateAtLeast(ctl.get(), STOP) ||
                 (Thread.interrupted() &&
                  runStateAtLeast(ctl.get(), STOP))) &&
                !wt.isInterrupted())
                wt.interrupt();
            try {
                beforeExecute(wt, task);
                Throwable thrown = null;
                try {
                    task.run();
                } catch (RuntimeException x) {
                    thrown = x; throw x;
                } catch (Error x) {
                    thrown = x; throw x;
                } catch (Throwable x) {
                    thrown = x; throw new Error(x);
                } finally {
                    afterExecute(task, thrown);
                }
            } finally {
                task = null;
                w.completedTasks++;
                w.unlock();
            }
        }
        completedAbruptly = false;
    } finally {
        // 对于非核心线程的线程达到keepAliveTime或允许核销线程超时时，回收超时线程
        processWorkerExit(w, completedAbruptly);
    }
}

这块代码的关键代码为while (task != null || (task = getTask()) != null)，作用为读取阻塞队列中的待执行任务。getTask()中Worker线程从阻塞获取元素，而ScheduledThreadPoolExecutor的DelayedWorkQueue在获取元素时，如果有任务的下一次执行时间到了则立即返回，否则通过互斥锁阻塞直到有新元素添加或者有元素的下一次执行时间到了。

因此Worker线程在while循环中不断阻塞-执行任务-阻塞，从而实现了循环间隔固定时间执行任务的效果。

>