Android Java 线程池 ScheduledThreadPoolExecutor源码篇

曾经终败给现在 2022-07-28 01:34 85阅读 0赞

接上前一篇文章[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]，我们继续来分析ScheduledThreadPoolExecutor源码的简单实现。

ScheduledThreadPoolExecutor可以添加定时任务的线程池，包括添加周期性定时任务。在前一篇文章[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]基础上来看这篇文件应该更加的方便一点。

ScheduledThreadPoolExecutor构造函数

public ScheduledThreadPoolExecutor(int corePoolSize) {
            super(corePoolSize, Integer.MAX_VALUE,
                  DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,
                  new DelayedWorkQueue());
        }

我们先关注两个东西DelayedWorkQueue 和DelayedWorkQueue 队列里面放的RunnableScheduledFuture(ScheduledFutureTask)  
先说DelayedWorkQueue 我们关心里面的offer, poll,take三个方法，DelayedWorkQueue 是以数组的形式存放RunnableScheduledFuture(这个类我们稍后分析)这个类实现了Comparable接口，在每次offer的时候会调用siftUp函数根据compareTo的比较值插入到队列中并且保证队列中的顺序是从小到大的顺序。数组下标低位的小，高位的大。poll的时候也是先从低位取出然后siftDown调整数组下标。其实compareTo比较的依据就是计算之后任务要跑的时间点下面会提到。

在一个就是RunnableScheduledFuture(ScheduledFutureTask) 一步一步的来看吧 那我们就直接看ScheduledFutureTask类图  
![这里写图片描述][20160329193545400]

先看下ScheduledFutureTask的成员变量，我们关注三个  
1. Callable 前一篇文章也有说过 队列里面放得是FutureTask现在放的是ScheduledFutureTask，当线程从队列里面把FutureTask(ScheduledFutureTask)拿出来之后调用FutureTask(ScheduledFutureTask)的run方法。run方法里面会调用FutureTask里面Callable的call方法，call方法调用完之后保存住了call的返回值。FutureTask 可以通过get方法得到这个返回值。  
2. time 这个是任务执行的时间(延时时间加当前时间计算来的)  
3. period 如果是周期任务，表示周期的间隔时间。如果大于0表示的是上一个任务开始到下一个任务开始的时间间隔(scheduleAtFixedRate)，如果是小于则表示上一个任务结束的时间到下一个任务开始的时间(scheduleWithFixedDelay)。

ScheduledFutureTask构造函数三个 有的传入的是Runnable 有的传入的是Callable，就算传入的是Runnable也会构建出一个Callable的对象，在Callable的call方法中调用Runnable的run方法。最后都是Callable。那这三个构造函数就没什么看的了。

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

再关注ScheduledFutureTask里面的getDelay，compareTo，isPeriodic，setNextRunTime几个方法。  
getDelay 方法简单，就是time减去当前时间。用来判断任务是不是该执行了。

public long getDelay(TimeUnit unit) {
            return unit.convert(time - now(), NANOSECONDS);
        }

compareTo 方法比较的就是time。在任务加入的时候用来找到新的任务在队列中的位置。

public int compareTo(Delayed other) {
            if (other == this) // compare zero if same object
                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;
        }

isPeriodic 任务是不是周期任务

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

setNextRunTime 如果是周期性的任务算出任务的下次执行时间，当周期性任务执行了第一次之后要算第二次的执行时间，依次类推。period > 0 上一个任务开始时间到下一个任务结束时间 time+=p(scheduleAtFixedRate), perid<0上一个任务结束时间到下一个任务开始时间那肯定是当前时间+ (-p)(scheduleWithFixedDelay) 这个函数是会在周期性任务执行完之后调用。

private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }

总结下RunnableScheduledFuture(ScheduledFutureTask) 和DelayedWorkQueue 部分为接下的的ScheduledThreadPoolExecutor做些准备哦。  
1. 入队offer的时候会按照任务RunnableScheduledFuture(ScheduledFutureTask)的time(任务的运行时间)来排序。从小到大来排序。  
2. 出队poll的时候先取出index 0位置上的任务(因为0位置的任务时间小最先执行)。然后再调整队列。  
3. ScheduledFutureTask 里面getDelay() 方法可以得到当前任务还有多长时间执行。  
4. ScheduledFutureTask 里面setNextRunTime()如果周期性任务执行完了一次通过他得到下一次的执行时间，计算完之后再重新放到队列里面去(稍后讲)  
5. ScheduledFutureTask 里面isPeriodic 当前任务是不是周期任务，和第四点配合使用。

ScheduledThreadPoolExecutor的队列和队列里面的任务我都清楚了一点。接下来就是ScheduledThreadPoolExecutor源码分析了  
和分析[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]时候一样。构造函数我们就不看了 看submit函数，不管是哪个submit函数 最后走的都是schedule函数。我们直接看schedule函数。

...
    
        public ScheduledFuture<?> schedule(Runnable command,
                                           long delay,
                                           TimeUnit unit) {
            if (command == null || unit == null)
                throw new NullPointerException();
            RunnableScheduledFuture<Void> t = decorateTask(command,
                                                           new ScheduledFutureTask<Void>(command, null,
                                                                                         triggerTime(delay, unit)));
            delayedExecute(t);
            return t;
        }
    
        public <V> ScheduledFuture<V> schedule(Callable<V> callable,
                                               long delay,
                                               TimeUnit unit) {
            if (callable == null || unit == null)
                throw new NullPointerException();
            RunnableScheduledFuture<V> t = decorateTask(callable,
                                                        new ScheduledFutureTask<V>(callable,
                                                                                   triggerTime(delay, unit)));
            delayedExecute(t);
            return t;
        }
    
        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();
            ScheduledFutureTask<Void> sft =
                new ScheduledFutureTask<Void>(command,
                                              null,
                                              triggerTime(initialDelay, unit),
                                              unit.toNanos(period));
            RunnableScheduledFuture<Void> t = decorateTask(command, sft);
            sft.outerTask = t;
            delayedExecute(t);
            return t;
        }
    
        public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command,
                                                         long initialDelay,
                                                         long delay,
                                                         TimeUnit unit) {
            if (command == null || unit == null)
                throw new NullPointerException();
            if (delay <= 0)
                throw new IllegalArgumentException();
            ScheduledFutureTask<Void> sft =
                new ScheduledFutureTask<Void>(command,
                                              null,
                                              triggerTime(initialDelay, unit),
                                              unit.toNanos(-delay));
            RunnableScheduledFuture<Void> t = decorateTask(command, sft);
            sft.outerTask = t;
            delayedExecute(t);
            return t;
        }
    
        ...

上面四个方法不管是哪个做的事情都差不多，都是先构造出RunnableScheduledFureture(ScheduledFutureTask)对象然后调用delayedExecutef(t)函数。我们就看下ScheduledFutureTask的构造过程。看到每个都调用了triggerTime(initialDelay, unit) 这个函数的就是通过delay的时间去计算出任务要执行的时间。  
scheduleAtFixedRate scheduleWithFixedDelay两个函数的区别主要看第三个参数的区别，scheduleAtFixedRate 表示上一个任务开始到下一个任务开始的时间。scheduleWithFixedDelay 则表示上一个任务结束到下一个任务开始的时间。从这里的代码里面我们就看到scheduleAtFixedRate 给ScheduledFutureTask里的period 是大于0的，scheduleWithFixedDelay 则是小于0的。 正好映射到ScheduledFutureTask 里面setNextRunTime()方法。  
继续往下走了delayedExecute 函数

private void delayedExecute(RunnableScheduledFuture<?> task) {
            if (isShutdown())
                reject(task);
            else {
                super.getQueue().add(task);
                if (isShutdown() &&
                    !canRunInCurrentRunState(task.isPeriodic()) &&
                    remove(task))
                    task.cancel(false);
                else
                    ensurePrestart();
            }
        }

第5行 super.getQueue().add(task); 接着会调用DelayedWorkQueue 的offer方法 前面也有说到入队offer的时候会按照任务RunnableScheduledFuture(ScheduledFutureTask)的time(任务的运行时间)来排序。从小到大来排序。  
第11行 ensurePrestart函数调用了addWorker()函数 addWorker和前一篇文章[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]里面的addWorker是一样。

void ensurePrestart() {
            int wc = workerCountOf(ctl.get());
            if (wc < corePoolSize)
                addWorker(null, true);
            else if (wc == 0)
                addWorker(null, false);
        }

接下来我们就该关心怎么从队列里面拿任务了(根据前文[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]的分析)，分析到上面线程已经启动起来了。从队列里面拿任务关心队列的take 和 poll方法，也就是DelayedWorkQueue 的take 和 poll方法。

先看DelayedWorkQueue 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);
                        if (delay <= 0)
                            return finishPoll(first);
                        first = null; // don't retain ref while waiting
                        if (leader != null)
                            available.await();
                        else {
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                available.awaitNanos(delay);
                            } finally {
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                if (leader == null && queue[0] != null)
                    available.signal();
                lock.unlock();
            }
        }

具体干的事情就是拿到index=0的任务。得到delay 如果delay<=0 这个任务就应该运行了 前面有说getDelay的作用(ScheduledFutureTask 里面getDelay() 方法可以得到当前任务还有多长时间执行。)，其他情况就得available.await();阻塞等待了。finishPoll是拿出一个任务之后做相应的调整。

接着是poll方法了 至于什么时候会调用poll方法什么时候会调用take方法可以参考[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]

public RunnableScheduledFuture<?> poll(long timeout, TimeUnit unit)
            throws InterruptedException {
            long nanos = unit.toNanos(timeout);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {
                    RunnableScheduledFuture<?> first = queue[0];
                    if (first == null) {
                        if (nanos <= 0)
                            return null;
                        else
                            nanos = available.awaitNanos(nanos);
                    } else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return finishPoll(first);
                        if (nanos <= 0)
                            return null;
                        first = null; // don't retain ref while waiting
                        if (nanos < delay || leader != null)
                            nanos = available.awaitNanos(nanos);
                        else {
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                long timeLeft = available.awaitNanos(delay);
                                nanos -= delay - timeLeft;
                            } finally {
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                if (leader == null && queue[0] != null)
                    available.signal();
                lock.unlock();
            }
        }

和take差不多，只是这个不会一直阻塞住有timeout时间的。这样一次任务的调用时间跑完了吧，结合上一篇文章[Android Java 线程池 ThreadPoolExecutor源码篇][Android Java _ ThreadPoolExecutor]的分析。

就剩下周期性的任务了。接着看下周期性的任务是怎么周期性执行的。看ScheduledFutureTask的run方法。

public void run() {
            boolean periodic = isPeriodic();
            if (!canRunInCurrentRunState(periodic))
                cancel(false);
            else if (!periodic)
                ScheduledFutureTask.super.run();
            else if (ScheduledFutureTask.super.runAndReset()) {
                setNextRunTime();
                reExecutePeriodic(outerTask);
            }
        }

不是周期性任务的时候直接调用run方法，这个好说，理解。不是周期性任务的时候先调用了runAndReset() 调用了ScheduledFutureTask 里面Callable的call方法 执行了任务的逻辑，又把任务的状态恢复了。  
第8行，setNextRunTime 前面有说吧目的是计算下次任务执行的时间。  
第9行，reExecutePeriodic 有把这个任务加入到队列里面去了。  
周期性任务也到此为止。

总结

1.  ScheduledThreadPoolExecutor 的队列是DelayedWorkQueue 在offer的时候会按照delay从小到大的顺序插入到队列当中去。poll 或者take的时候都是先拿的index=0的任务，然后看是否任务的getDelay方法是不是小于等于0，任务是否到了该执行的时间了。
2.  ScheduledThreadPoolExecutor 队列里面的任务是ScheduledFutureTask 实现了Comparable接口，在任务入队offer的时候用到。关键方法getDelay，compareTo，isPeriodic，setNextRunTime。
3.  周期性任务的执行都是每次任务执行完之后再计算出下一次的运行时间，然后再重新插入到队列中。在ScheduledFutureTask 的run方法中完成。

Executors 里面一些常用方法的介绍

1.  public static ExecutorService newFixedThreadPool(int nThreads)  
    固定nThreads线程个数的线程池。开始的时候当我们的任务的个数小于nThreads的时候线程会一个一个的开起来，当达到nThreads的时候这些线程就一直存在了。
2.  public static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory)  
    和前面是一样的值是自定义了线程工厂类，更加的灵活一点。
3.  public static ExecutorService newSingleThreadExecutor()  
    固定一个线程的线程池。
4.  public static ExecutorService newCachedThreadPool()  
    不规定线程的个数，当线程空闲时间大约60S的时候线程会关掉。
5.  public static ScheduledExecutorService newScheduledThreadPool(int corePoolSize)  
    核心线程为corePoolSize的周期性线程池。

Executors 的其他的一些static方法可能是多传入了一个ThreadFactory或者RejectedExecutionHandler就没有列出来了。

[Android Java _ ThreadPoolExecutor]: http://blog.csdn.net/wuyuxing24/article/details/50989530
[20160329193545400]: /images/20220727/80df4662e70f4d9fba08654d23ae1ab1.png