【Java 线程池】【八】ScheduledThreadPoolExecutor之ScheduledFutureTask、DelayedWorkQueue原理-526互联

1 前言

上一节我们看了ScheduledThreadPoolExecutor线程池提交任务的流程。execute、submit、schedule、scheduleAtFixRate方法的原理：都是将任务封装为一个ScheduledFutureTask，然后提交到延迟阻塞队列里面，然后线程池里的线程从延迟阻塞队列中获取到任务，然后执行。那么我们这节就来看看ScheduledFutureTask的原理。

2 ScheduledFutureTask 源码

2.1 ScheduledFutureTask 类结构图

（1）首先继承了FutureTask类，具有FutureTask的功能，至于FutureTask我们之前已经深入分析过了
（2）实现了Delayed接口，具有一个延迟时间属性，存放进入延迟队列的元素必须具有一个延迟时间属性，所以这里实现Delayed接口是必要的

public interface Delayed extends Comparable<Delayed> {
    // Delayed的getDelay接口，返回当前任务或者元素的延迟时间，也就是需要延迟多久
    long getDelay(TimeUnit unit);
}

（3）实现了RunnableScheduledFuture接口，用户判断这个任务是否是周期性任务。

public interface RunnableScheduledFuture<V> extends RunnableFuture<V>, ScheduledFuture<V> {
    // RunnableScheduledFuture的isPeriodic方法返回这个任务是否是周期性任务
    // true，周期性任务
    // false，非周期性任务
    boolean isPeriodic();
}

2.2 核心属性

// 序号id
// 默认情况下延迟任务按照时间优先级执行
// 也就是延迟时间越小，越先执行
// 但是如果两个任务延迟时间一样，则sequenceNumber小的优先级高，先执行
private final long sequenceNumber;

// 延迟时间，任务多久之后会执行
private long time;

// 是否是周期性重复任务
// 是否是0，则不是周期性任务，只是一次性的
// 如果大于0，则每隔period时间会重复执行一次
// 如果小于0，则每隔-period时间重复执行一次
private final long period;

// 这里就是具体的要执行的任务了
RunnableScheduledFuture<V> outerTask = this;

// 在延迟队列中的位置，DelyedWorkQueue延迟队列是基于数组和堆排序实现的
// 而ScheduledFutureTask存储在DelayedWorkQueue延迟队列中，也就是存在数组中
// 这里就是说这个任务是在堆的哪个位置，数组的哪个位置
int heapIndex;

2.3 构造方法

ScheduledFutureTask(Runnable r, V result, long ns) {
    // 调用父类FutureTask的构造方法
    super(r, result);
    // 具体的延迟时间
    this.time = ns;
    // 是否周期性任务，如果为0不是
    this.period = 0;
    // 任务的序号
    // 如果两个任务延时时间一样，需要小的有限执行
    this.sequenceNumber = sequencer.getAndIncrement();
}

ScheduledFutureTask(Runnable r, V result, long ns, long period) {
    super(r, result);
    this.time = ns;
    this.period = period;
    this.sequenceNumber = sequencer.getAndIncrement();
}

2.4 getDelay 方法

实现了Delayed接口的getDelay方法，对外提供一个入口获取任务的延迟时间：

public long getDelay(TimeUnit unit) {
    // 任务的延迟时间就是time - 当前时间，然后转成纳秒单位
    return unit.convert(time - now(), NANOSECONDS);
}

2.5 isPeriodic 方法

实现了RunnableScheduleFuture的isPeriodic接口，判断是否是周期性重复任务：

public boolean isPeriodic() {
    return period != 0;
}

2.6 setNextRunTime 方法

setNextRunTime 如果是周期性重复任务，设置下一次执行的时间：

private void setNextRunTime() {
    long p = period;
    if (p > 0)
        // 下一个周期执行时间为 time+p
        time += p;
    else
        // 下一个周期执行时间为 now() - p
        time = triggerTime(-p);
}

long triggerTime(long delay) {
    return now() +
        ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
}

2.7 compareTo 方法

任务的优先级比较，先通过延时时间比较，延迟时间小的优先级高，如果延迟时间一样，sequenceNumber小的优先级高：

public int compareTo(Delayed other) {
    if (other == this) // compare zero if same object
        return 0;
     // 如果Delayed延迟任务是SheduledFutureTask的子类
    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
        long diff = time - x.time;
        // 则先通过延迟时间比较，延迟时间小的优先级高
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        // 延迟时间一样则sequenceNumber小的优先级高
        else if (sequenceNumber < x.sequenceNumber)
            return -1;
        else
            return 1;
    }
    // 如果不是ScheduleFutureTask类的实例，直接比较延迟时间就好了
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}

2.8 run 方法

运行任务的run方法：

public void run() {
    // 首先获取任务是否是周期性重复任务
    boolean periodic = isPeriodic();
    // 如果当前任务线程池状态不对，取消任务执行
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    // 如果不是周期性的任务，只是一次性延迟任务
    else if (!periodic)
        // 直接调用父类FutureTask的run方法运行任务
        ScheduledFutureTask.super.run();
    // 走到这里periodic为true，说明是周期性重复任务
    else if (ScheduledFutureTask.super.runAndReset()) {
        // 设置下一次运行的时间
        setNextRunTime();
        // 这里就是运行任务，然后将任务重新方法延迟队列中，因为下一次运行的时间修改了
        reExecutePeriodic(outerTask);
    }
}

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    if (canRunInCurrentRunState(true)) {
        // 再次放入延迟阻塞队列中
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

看下来重要的可能就是：

（1）ScheduleFutureTask通过实现了Delay接口，具有延迟时间的属性，通过实现了RunnableScheduleFuture接口具有判断是否是周期性重复任务
（2）通过继承了FutureTask具有Future的特性，可以获取任务的结果，对任务进行取消，获取任务的执行状态等。
（3）内部的compareTo为排序的比较方法，首先按照延迟时间进行比较，延迟时间小的优先级高，如果延迟时间一样，那么按照sequenceNumber 比较，越小优先级越高。

3 DelayedWorkQueue 源码

看完ScheduleFutureTask，我们接着继续，看看DelayedWorkQueue是如何实现延迟阻塞队列的。

3.1 接口声明

// DelayedWorkQueue实现了BlockingQueue接口，是一个阻塞队列
static class DelayedWorkQueue extends AbstractQueue<Runnable>
    implements BlockingQueue<Runnable> {
}

3.2 内部属性

// 初始化容量
private static final int INITIAL_CAPACITY = 16;
// 这里就是阻塞队列内部存储数据的结构，是用一个数组来进行存储的
// 数组的初始化长度是16
private RunnableScheduledFuture<?>[] queue =
    new RunnableScheduledFuture<?>[INITIAL_CAPACITY];
// 互斥锁，为实现线程安全和阻塞
private final ReentrantLock lock = new ReentrantLock();
// 等待条件
private final Condition available = lock.newCondition();
// 阻塞队列的大小
private int size = 0;
// leader，第一个等待的线程
private Thread leader = null;

3.3 put、add 方法

put、add底层都是调用offer方法：

public void put(Runnable e) {
    offer(e);
}

public boolean add(Runnable e) {
    return offer(e);
}

public boolean offer(Runnable e, long timeout, TimeUnit unit) {
    return offer(e);
}

3.4 offer 方法

public boolean offer(Runnable x) {
    if (x == null)
        throw new NullPointerException();
    RunnableScheduledFuture<?> e = (RunnableScheduledFuture<?>)x;
    final ReentrantLock lock = this.lock;
    // 进行加锁
    lock.lock();
    try {
        // 获取当前阻塞队列的大小
        int i = size;
        // 如果当前队列大小大于等于内部数组的长度，说明存储任务的数组满了，需要进行扩容
        if (i >= queue.length)
            // 具体扩容的方法
            grow();
        // 队列大小+1
        size = i + 1;
        // 如果插入前i==0,说明自己是第一个插入元素，直接放入数组的0号元素即可
        if (i == 0) {
            queue[0] = e;
            // 设置当前任务e在数组中的位置
            setIndex(e, 0);
        } else {
            // 如果 i!=0 说明插入队列非空，数组有元素
            // 需要插入任务e，同时进行排序，延迟时间小的排在前面
            // 这里的siftUp方法就是插入和排序
            siftUp(i, e);
        }
        // 如果调整之后，刚插入的元素就是延迟时间最小的元素，唤醒沉睡线程
        if (queue[0] == e) {
            leader = null;
            available.signal();
        }
    } finally {
        // 释放锁
        lock.unlock();
    }
    return true;
}

3.4.1 grow 数组扩容方法

private void grow() {
    // 获取旧数组长度
    int oldCapacity = queue.length;
    // 新数组长度为旧数组长度的1.5倍
    // 比如oldCapacity = 10
    // newCapacity = 10 + (10 > 1) = 10 + 5 = 15
    int newCapacity = oldCapacity + (oldCapacity >> 1); // grow 50%
    // 如果新数组长度小于0，那么新数组长度MAX_VALUE
    if (newCapacity < 0) // overflow
        newCapacity = Integer.MAX_VALUE;
    // 这里就是将旧数组元素拷贝到新数组了，没啥好说的
    queue = Arrays.copyOf(queue, newCapacity);
}

3.4.2 siftUp 插入和排序方法

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

上面的方法，其实就是：
（1）新插入一个元素放入数组尾部
（2）然后通过堆排序，不断调整，形成一个小顶堆

3.5 poll 非阻塞方法

public RunnableScheduledFuture<?> poll() {
    final ReentrantLock lock = this.lock;
    // 获取锁
    lock.lock();
    try {
        // 获取小顶堆的根元素
        RunnableScheduledFuture<?> first = queue[0];
        // 如果为null说明队列时空
        // 如果延迟时间大于当前试下，说明任务还没到执行的时候
        if (first == null || first.getDelay(NANOSECONDS) > 0)
            // 当前没有可以执行的任务，返回null
            return null;
        else
            // 走到这里fisrt非null，并且延迟时间小于当前时间，说明可以执行了
            //
            return finishPoll(first);
    } finally {
        // 释放锁
        lock.unlock();
    }
}

3.5.1 finishPoll 方法

private RunnableScheduledFuture<?> finishPoll(RunnableScheduledFuture<?> f) {
    // 队列出队一个元素，大小减少1
    int s = --size;
    RunnableScheduledFuture<?> x = queue[s];
    queue[s] = null;
    // 重新调整堆结构，调整成小顶堆
    if (s != 0)
        siftDown(0, x);
    setIndex(f, -1);
    return f;
}

3.6 take 阻塞方法

public RunnableScheduledFuture<?> take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        // 循环重试
        for (;;) {
            RunnableScheduledFuture<?> first = queue[0];
            // 如果当前队列没有元素
            if (first == null)
                // avaiable 可用条件不满足
                // 调用await方法释放锁，同时进入条件队列进行沉睡等待
                // 等待当队列有元素的时候，调用singal方法唤醒自己
                available.await();
            else {
                
                long delay = first.getDelay(NANOSECONDS);
                // 当前有元素，并且延迟时间小于当前时间，符合执行条件，完成出队
                if (delay <= 0)
                    return finishPoll(first);
                // 走到这里，说明delay > 0 队列中的元素延迟执行时间都大于当前时间
                // 没有可以执行的任务
                first = null; // don't retain ref while waiting
                // 这里leader表示第一个获取元素失败则阻塞的线程
                // 如果leader不为null，前面已经有人等着了，自己只需要等待就可以，别人会唤醒自己
                if (leader != null)
                    // 进入等待队列等待
                    available.await();
                else {
                    // 走到这里说明leader == null
                    Thread thisThread = Thread.currentThread();
                    leader = thisThread;
                    try {
                        // 由于阻塞队列第一个元素至少还有delay时间的延迟
                        // 所以自己只需要等待delay时间，就可以获取到元素
                        // 所以这里就是阻塞等待delay时间
                        available.awaitNanos(delay);
                    } finally {
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }
    } finally {
        // 当前还有元素，继续唤醒其它等待线程，叫醒别人来取任务了
        if (leader == null && queue[0] != null)
            available.signal();
        // 释放锁
        lock.unlock();
    }
}