Java中的ScheduledThreadPoolExecutor定时任务详解

ScheduledThreadPoolExecutor详解

ScheduledThreadPoolExecutor 继承自 ThreadPoolExecutor。它主要用来在给定的延迟之后运行任务，或者定期执行任务。

ScheduledThreadPoolExecutor 的功能与 Timer 类似， 但 ScheduledThreadPoolExecutor 功能更强大、更灵活。

Timer 对应的是单个后台线程，而 ScheduledThreadPoolExecutor 可以在构造函数中指定多个对应的后台线程数。

ScheduledThreadPoolExecutor 的运行机制

ScheduledThreadPoolExecutor 的任务传递示意图

ScheduledThreadPoolExecutor 的执行主要分为两大部分：

1) 当调用 ScheduledThreadPoolExecutor 的 scheduleAtFixedRate()方法或者 scheduleWithFixedDelay()方法时，会向 ScheduledThreadPoolExecutor 的 DelayQueue 添加一个实现了 RunnableScheduledFutur 接口的 ScheduledFutureTask。

2) 线程池中的线程从 DelayQueue 中获取 ScheduledFutureTask，然后执行任务。

ScheduledThreadPoolExecutor 为了实现周期性的执行任务，对 ThreadPoolExecutor 做 了如下的修改：

•  使用 DelayQueue 作为任务队列。
• 获取任务的方式不同。
• 执行周期任务后，增加了额外的处理。

ScheduledThreadPoolExecutor 的实现

ScheduledThreadPoolExecutor 会把待调度的任务 （ScheduledFutureTask）放到一个 DelayQueue 中。

ScheduledFutureTask 主要包含 3 个成员变量，如下。

• long 型成员变量 time，表示这个任务将要被执行的具体时间。
• long 型成员变量 sequenceNumber，表示这个任务被添加到 ScheduledThreadPoolExecutor 中的序号。
• long 型成员变量 period，表示任务执行的间隔周期。

DelayQueue 封装了一个 PriorityQueue，这个 PriorityQueue 会对队列中的 ScheduledFutureTask 进行排序。排序时，time 小的排在前面（时间早的任务将被先执行）。

如果两个 ScheduledFutureTask 的 time 相同，就比较 sequenceNumber，sequenceNumber 小的排在前面（也就是说，如果两个任务的执行时间相同，那么先提交的任务将被先执行）。

ScheduledThreadPoolExecutor 中的线程执行周期任务的过程

线程 1 从 DelayQueue 中获取已到期的 ScheduledFutureTask（DelayQueue.take()）。

到 期任务是指 ScheduledFutureTask 的 time 大于等于当前时间。线程 1 执行这个 ScheduledFutureTask。

线程 1 修改 ScheduledFutureTask 的 time 变量为下次将要被执行的时间。

线程 1 把这个修改 time 之后的 ScheduledFutureTask 放回 DelayQueue 中（DelayQueue.add()）。

DelayQueue.take()方法

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly(); // 1  获取响应打断的锁
    try {
        for (; ; ) {
            E first = q.peek(); // 获取ScheduledFutureTask 
            if (first == null) {
                available.await(); // 2.1 到 Condition 中等待
            } else {
                long delay = first.getDelay(TimeUnit.NANOSECONDS);
                if (delay > 0) {
                    // 2.2 到 Condition 中等待到 time 时间
                    long tl = available.awaitNanos(delay); 
                } else {
                    E x = q.poll(); // 2.3.1 获取 PriorityQueue 的头元素
                    assert x != null;
                    if (q.size() != 0)
                        available.signalAll(); // 2.3.2 唤醒在 Condition 中等待的所有线程
                    return x;
                }
            }
        }
    } finally {
        lock.unlock(); // 3 释放🔒
    }
}

1) 获取 Lock。

2) 获取周期任务。

• 如果 PriorityQueue 为空，当前线程到 Condition 中等待；否则执行下面的 2.2。
• 如果 PriorityQueue 的头元素的 time 时间比当前时间大，到 Condition 中等待到 time 时间；否则执行下面的 2.3。
• 获取 PriorityQueue 的头元素（2.3.1）；如果 PriorityQueue 不为空，则唤醒在 Condition 中等待的所有线程（2.3.2）。

3) 释放 Lock。

ScheduledThreadPoolExecutor 在一个循环中执行步骤 2，直到线程从 PriorityQueue 获 取到一个元素之后（执行 2.3.1 之后），才会退出无限循环（结束步骤 2）。

DelayQueue.offer()方法

// 把 ScheduledFutureTask 放入 DelayQueue 中的过程
public boolean offer(E e) {
    final ReentrantLock lock = this.lock;
    lock.lock(); // 1
    try {
        E first = q.peek();
        q.offer(e); // 2.1
        if (first == null || e.compareTo(first) < 0) available.signalAll(); // 2.2
        return true;
    } finally {
        lock.unlock(); // 3
    }
}

1) 获取 Lock。

2) 添加任务。

• 向 PriorityQueue 添加任务。
• 如果在上面 2.1 中添加的任务是 PriorityQueue 的头元素，唤醒在 Condition 中等 待的所有线程。

3) 释放 Lock

到此这篇关于Java中的ScheduledThreadPoolExecutor定时任务详解的文章就介绍到这了,更多相关ScheduledThreadPoolExecutor详解内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

最新评论