高并发之——深度解析ScheduledFutureTask类源码

冰河

发布于 2020-10-29 16:01:20

4460

发布于 2020-10-29 16:01:20

文章被收录于专栏：冰河技术

作者个人研发的在高并发场景下，提供的简单、稳定、可扩展的延迟消息队列框架，具有精准的定时任务和延迟队列处理功能。自开源半年多以来，已成功为十几家中小型企业提供了精准定时调度方案，经受住了生产环境的考验。为使更多童鞋受益，现给出开源框架地址：

https://github.com/sunshinelyz/mykit-delay

PS: 欢迎各位Star源码，也可以pr你牛逼哄哄的代码。

JDK 1.5提供的ScheduledThreadPoolExecutor执行定时任务时，会将任务封装成ScheduledFutureTask对象。那么，ScheduledFutureTask对象有何特殊之处？今天，我们就一起来手撕ScheduledFutureTask类的源码，来深入理解ScheduledFutureTask类的细节。

类的层级关系

从ScheduledFutureTask类的定义可以看出，ScheduledFutureTask类是ScheduledThreadPoolExecutor类的私有内部类，继承了FutureTask类，并实现了RunnableScheduledFuture接口。也就是说，ScheduledFutureTask具有FutureTask类的所有功能，并实现了RunnableScheduledFuture接口的所有方法。ScheduledFutureTask类的定义如下所示。

private class ScheduledFutureTask<V> extends FutureTask<V> implements RunnableScheduledFuture<V>

我们也可以从ScheduledFutureTask类的层级关系图，更加清晰的看出ScheduledFutureTask类继承了哪些类和实现了哪些接口，有关ScheduledFutureTask类的层级关系图如下所示。

接下来，我们就拆解ScheduledFutureTask类的源码，探究其细节信息。我们还是带着一些问题来解读ScheduledFutureTask类的源码，例如为何将任务封装成ScheduledFutureTask对象？在定时任务中，如何区分不同的 ScheduledFutureTask任务？

重要的成员变量

ScheduledFutureTask类中提供了几个成员变量，如下所示。

//任务添加到ScheduledThreadPoolExecutor中被分配的唯一序列号
private final long sequenceNumber;
//下次执行任务的时间
private long time;
//任务执行的周期
private final long period;
//ScheduledFutureTask对象，实际指向当前对象本身
RunnableScheduledFuture<V> outerTask = this;
//当前任务在延迟队列中的索引
//能够更加方便的取消当前任务
int heapIndex;

我们来看几个重要的属性。

sequenceNumber：任务添加到ScheduledThreadPoolExecutor中被分配的唯一序列号，可以根据这个序列号确定唯一的一个任务，如果在定时任务中，如果一个任务是周期性执行的，但是它们的sequenceNumber的值相同，则被视为是同一个任务。
time：下一次执行任务的时间。
period：任务的执行周期。
outerTask：ScheduledFutureTask对象，实际指向当前对象本身。此对象的引用会被传入到周期性执行任务的ScheduledThreadPoolExecutor类的reExecutePeriodic方法中。
heapIndex：当前任务在延迟队列中的索引，这个索引能够更加方便的取消当前任务。

构造方法

ScheduledFutureTask类继承了FutureTask类，并实现了RunnableScheduledFuture接口。在ScheduledFutureTask类中提供了如下构造方法。

ScheduledFutureTask(Runnable r, V result, long ns) {
    super(r, result);
    this.time = ns;
    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();
}

ScheduledFutureTask(Callable<V> callable, long ns) {
    super(callable);
    this.time = ns;
    this.period = 0;
    this.sequenceNumber = sequencer.getAndIncrement();
}

通过源码可以看到，在ScheduledFutureTask类的构造方法中，首先会调用FutureTask类的构造方法为FutureTask类的callable和state成员变量赋值，接下来为ScheduledFutureTask类的time、period和sequenceNumber成员变量赋值。理解起来比较简单。

getDelay方法

我们先来看getDelay方法的源码，如下所示。

//获取下次执行任务的时间距离当前时间的纳秒数
public long getDelay(TimeUnit unit) {
    return unit.convert(time - now(), NANOSECONDS);
}

getDelay方法比较简单，主要用来获取下次执行任务的时间距离当前系统时间的纳秒数。

compareTo方法

ScheduledFutureTask类在类的结构上实现了Comparable接口，compareTo方法主要是对Comparable接口定义的compareTo方法的实现。源码如下所示。

public int compareTo(Delayed other) {
    if (other == this) 
        return 0;
    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask<?> x = (ScheduledFutureTask<?>)other;
        long diff = time - x.time;
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        else if (sequenceNumber < x.sequenceNumber)
            return -1;
        else
            return 1;
    }
    long diff = getDelay(NANOSECONDS) - other.getDelay(NANOSECONDS);
    return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
}

这段代码看上去好像是对各种数值类型数据的比较，本质上是对延迟队列中的任务进行排序。排序规则为：首先比较延迟队列中每个任务下次执行的时间，下次执行时间距离当前时间短的任务会排在前面；如果下次执行任务的时间相同，则会比较任务的sequenceNumber值，sequenceNumber值小的任务会排在前面。

isPeriodic方法

isPeriodic方法的源代码如下所示。

//判断是否是周期性任务
public boolean isPeriodic() {
    return period != 0;
}

这个方法的作用比较简单，主要是用来判断当前任务是否是周期性任务。这里只要判断运行任务的执行周期不等于0就能确定为周期性任务了。因为无论period的值是大于0还是小于0，当前任务都是周期性任务。

setNextRunTime方法

setNextRunTime方法的作用主要是设置当前任务下次执行的时间，源码如下所示。

private void setNextRunTime() {
    long p = period;
    //固定频率，上次执行任务的时间加上任务的执行周期
    if (p > 0)
        time += p;
    //相对固定的延迟执行，当前系统时间加上任务的执行周期
    else
        time = triggerTime(-p);
}

这里，再一次证明了使用isPeriodic方法判断当前任务是否为周期性任务时，只要判断period的值是否不等于0就可以了。因为如果当前任务是固定频率执行的周期性任务，会将周期period当作正数来处理；如果是相对固定的延迟执行当前任务，则会将周期period当作负数来处理。

这里，我们看到在setNextRunTime方法中，调用了ScheduledThreadPoolExecutor类的triggerTime方法。接下来，我们看下triggerTime方法的源码。

ScheduledThreadPoolExecutor类的triggerTime方法

triggerTime方法用于获取延迟队列中的任务下一次执行的具体时间。源码如下所示。

private long triggerTime(long delay, TimeUnit unit) {
    return triggerTime(unit.toNanos((delay < 0) ? 0 : delay));
}

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

这两个triggerTime方法的代码比较简单，就是获取下一次执行任务的具体时间。有一点需要注意的是：delay < (Long.MAX_VALUE >> 1判断delay的值是否小于Long.MAX_VALUE的一半，如果小于Long.MAX_VALUE值的一半，则直接返回delay，否则需要处理溢出的情况。

我们看到在triggerTime方法中处理防止溢出的逻辑使用了ScheduledThreadPoolExecutor类的overflowFree方法，接下来，我们就看看ScheduledThreadPoolExecutor类的overflowFree方法的实现。

ScheduledThreadPoolExecutor类的overflowFree方法

overflowFree方法的源代码如下所示。

private long overflowFree(long delay) {
    //获取队列中的节点
    Delayed head = (Delayed) super.getQueue().peek();
    //获取的节点不为空，则进行后续处理
    if (head != null) {
        //从队列节点中获取延迟时间
        long headDelay = head.getDelay(NANOSECONDS);
        //如果从队列中获取的延迟时间小于0，并且传递的delay
        //值减去从队列节点中获取延迟时间小于0
        if (headDelay < 0 && (delay - headDelay < 0))
            //将delay的值设置为Long.MAX_VALUE + headDelay
            delay = Long.MAX_VALUE + headDelay;
    }
    //返回延迟时间
    return delay;
}

通过对overflowFree方法的源码分析，可以看出overflowFree方法本质上就是为了限制队列中的所有节点的延迟时间在Long.MAX_VALUE值之内，防止在compareTo方法中溢出。

cancel方法

cancel方法的作用主要是取消当前任务的执行，源码如下所示。

public boolean cancel(boolean mayInterruptIfRunning) {
    //取消任务，返回任务是否取消的标识
    boolean cancelled = super.cancel(mayInterruptIfRunning);
    //如果任务已经取消
    //并且需要将任务从延迟队列中删除
    //并且任务在延迟队列中的索引大于或者等于0
    if (cancelled && removeOnCancel && heapIndex >= 0)
        //将当前任务从延迟队列中删除
        remove(this);
    //返回是否成功取消任务的标识
    return cancelled;
}

这段代码理解起来相对比较简单，首先调用取消任务的方法，并返回任务是否已经取消的标识。如果任务已经取消，并且需要移除任务，同时，任务在延迟队列中的索引大于或者等于0，则将当前任务从延迟队列中移除。最后返回任务是否成功取消的标识。

run方法

run方法可以说是ScheduledFutureTask类的核心方法，是对Runnable接口的实现，源码如下所示。

public void run() {
    //当前任务是否是周期性任务
    boolean periodic = isPeriodic();
    //线程池当前运行状态下不能执行周期性任务
    if (!canRunInCurrentRunState(periodic))
        //取消任务的执行
        cancel(false);
    //如果不是周期性任务
    else if (!periodic)
        //则直接调用FutureTask类的run方法执行任务
        ScheduledFutureTask.super.run();
    //如果是周期性任务，则调用FutureTask类的runAndReset方法执行任务
    //如果任务执行成功
    else if (ScheduledFutureTask.super.runAndReset()) {
        //设置下次执行任务的时间
        setNextRunTime();
        //重复执行任务
        reExecutePeriodic(outerTask);
    }
}

整个方法的逻辑就是：首先判断当前任务是否是周期性任务。如果线程池当前运行状态下不能执行周期性任务，则取消任务的执行。如果当前任务不是周期性任务，则直接调用FutureTask类的run方法执行任务；如果当前任务是周期性任务，则调用FutureTask类的runAndReset方法执行任务，并且如果任务执行成功，则设置下次执行任务的时间，同时，将任务设置为重复执行。

这里，调用了FutureTask类的run方法和runAndReset方法，并且调用了ScheduledThreadPoolExecutor类的reExecutePeriodic方法。接下来，我们分别看下这些方法的实现。

FutureTask类的run方法

FutureTask类的run方法源码如下所示。

public void run() {
    //状态如果不是NEW，说明任务或者已经执行过，或者已经被取消，直接返回
    //状态如果是NEW，则尝试把当前执行线程保存在runner字段中
    //如果赋值失败则直接返回
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset, null, Thread.currentThread()))
        return;
    try {
        Callable<V> c = callable;
        if (c != null && state == NEW) {
            V result;
            boolean ran;
            try {
                //执行任务
                result = c.call();
                ran = true;
            } catch (Throwable ex) {
                result = null;
                ran = false;
                //任务异常
                setException(ex);
            }
            if (ran)
                //任务正常执行完毕
                set(result);
        }
    } finally {

        runner = null;
        int s = state;
        //如果任务被中断，执行中断处理
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
}

代码的整体逻辑为：判断当前任务的state是否等于NEW,如果不为NEW则说明任务或者已经执行过，或者已经被取消，直接返回；如果状态为NEW则接着会通过unsafe类把任务执行线程引用CAS的保存在runner字段中，如果保存失败，则直接返回；执行任务；如果任务执行发生异常，则调用setException()方法保存异常信息。

FutureTask类的runAndReset方法

方法的源码如下所示。

protected boolean runAndReset() {
    if (state != NEW ||
        !UNSAFE.compareAndSwapObject(this, runnerOffset,
                                     null, Thread.currentThread()))
        return false;
    boolean ran = false;
    int s = state;
    try {
        Callable<V> c = callable;
        if (c != null && s == NEW) {
            try {
                c.call(); // don't set result
                ran = true;
            } catch (Throwable ex) {
                setException(ex);
            }
        }
    } finally {
        // runner must be non-null until state is settled to
        // prevent concurrent calls to run()
        runner = null;
        // state must be re-read after nulling runner to prevent
        // leaked interrupts
        s = state;
        if (s >= INTERRUPTING)
            handlePossibleCancellationInterrupt(s);
    }
    return ran && s == NEW;
}

FutureTask类的runAndReset方法与run方法的逻辑基本相同，只是runAndReset方法会重置当前任务的执行状态。

ScheduledThreadPoolExecutor类的reExecutePeriodic方法

ScheduledThreadPoolExecutor类的reExecutePeriodic方法的源代码如下所示。

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
    //线程池当前状态下能够执行任务
    if (canRunInCurrentRunState(true)) {
        //将任务放入队列
        super.getQueue().add(task);
        //线程池当前状态下不能执行任务，并且成功移除任务
        if (!canRunInCurrentRunState(true) && remove(task))
            //取消任务
            task.cancel(false);
        else
            //调用ThreadPoolExecutor类的ensurePrestart()方法
            ensurePrestart();
    }
}

总体来说reExecutePeriodic方法的逻辑比较简单，需要注意的是：调用reExecutePeriodic方法的时候已经执行过一次任务，所以，并不会触发线程池的拒绝策略；传入reExecutePeriodic方法的任务一定是周期性的任务。

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2020-03-02，如有侵权请联系 cloudcommunity@tencent.com 删除

linux

本文分享自冰河技术微信公众号，前往查看

如有侵权，请联系 cloudcommunity@tencent.com 删除。

本文参与腾讯云自媒体同步曝光计划，欢迎热爱写作的你一起参与！

linux

登录后参与评论

0 条评论

热度