ScheduledThreadPoolExecutor

红太狼 2022-09-13 11:26 149阅读 0赞

ScheduledThreadPoolExecutor可以按时间来执行任务，它继承了ThreadPoolExecutor

构造方法

public ScheduledThreadPoolExecutor(int corePoolSize) {
        
        	//使用ThreadPoolExecutor的构造方法
        	//DelayedWorkQueue是ScheduledThreadPoolExecutor内部自己实现的阻塞队列
            super(corePoolSize, Integer.MAX_VALUE, 0, NANOSECONDS,
                  new DelayedWorkQueue());
        }

#### 延迟执行 ####

public ScheduledFuture<?> schedule(Runnable command,
                                           long delay,
                                           TimeUnit unit) {
        
            //参数校验
            if (command == null || unit == null)
                throw new NullPointerException();
            //把触发时间和任务command封装成RunnableScheduledFuture
            RunnableScheduledFuture<?> t = decorateTask(command,
                new ScheduledFutureTask<Void>(command, null,
                				//根据延迟时间和TimeUnit计算出，任务触发的时间戳
                                              triggerTime(delay, unit)));
            //延迟执行任务
            delayedExecute(t);
            return t;
        }

private void delayedExecute(RunnableScheduledFuture<?> task) {
        
        	//如果线程池已经关闭了，执行拒绝策略
            if (isShutdown())
                reject(task);
            else {
        
            	//把task放进DelayedWorkQueue二叉堆中，唤醒因为available阻塞的线程
                super.getQueue().add(task);
                if (isShutdown() &&
                    !canRunInCurrentRunState(task.isPeriodic()) &&
                    remove(task))
                    task.cancel(false);
                else
                	//执行任务
                    ensurePrestart();
            }
        }
    
        void ensurePrestart() {
        
            int wc = workerCountOf(ctl.get());
            if (wc < corePoolSize)
                addWorker(null, true);
            else if (wc == 0)
                addWorker(null, false);
        }

进入父类ThreadPoolExecutor的addWorker方法

//传入的firstTask为null
    private boolean addWorker(Runnable firstTask, boolean core) {
        
            retry:
            for (;;) {
        
            	//检查线程池的状态
                int c = ctl.get();
                int rs = runStateOf(c);
    
                // Check if queue empty only if necessary.
                if (rs >= SHUTDOWN &&
                    ! (rs == SHUTDOWN &&
                       firstTask == null &&
                       ! workQueue.isEmpty()))
                    return false;
    			//检查执行worker的数量
                for (;;) {
        
                    int wc = workerCountOf(c);
                    if (wc >= CAPACITY ||
                        wc >= (core ? corePoolSize : maximumPoolSize))
                        return false;
                    if (compareAndIncrementWorkerCount(c))
                        break retry;
                    c = ctl.get();  // Re-read ctl
                    if (runStateOf(c) != rs)
                        continue retry;
                    // else CAS failed due to workerCount change; retry inner loop
                }
            }
    
            boolean workerStarted = false;
            boolean workerAdded = false;
            Worker w = null;
            try {
        
            	//包装成一个worker,firstTask是空的,worker实现了Runnable接口
                w = new Worker(firstTask);
                final Thread t = w.thread;
                if (t != null) {
        
                    final ReentrantLock mainLock = this.mainLock;
                    mainLock.lock();
                    try {
        
                    	//检查线程池的状态
                        int rs = runStateOf(ctl.get());
    
                        if (rs < SHUTDOWN ||
                            (rs == SHUTDOWN && firstTask == null)) {
        
                            if (t.isAlive()) // precheck that t is startable
                                throw new IllegalThreadStateException();
                            //把worker加入workers列表中
                            workers.add(w);
                            int s = workers.size();
                            if (s > largestPoolSize)
                                largestPoolSize = s;
                            workerAdded = true;
                        }
                    } finally {
        
                        mainLock.unlock();
                    }
                    //加入成功，执行worker的run方法
                    if (workerAdded) {
        
                    	//Worker的thread的start方法执行的是worker的run方法
                        t.start();
                        workerStarted = true;
                    }
                }
            } finally {
        
                if (! workerStarted)
                    addWorkerFailed(w);
            }
            return workerStarted;
        }

进入worker的run方法

public void run() {
        
                runWorker(this);
            }
        final void runWorker(Worker w) {
        
            Thread wt = Thread.currentThread();
            Runnable task = w.firstTask;
            w.firstTask = null;
            w.unlock(); // allow interrupts
            boolean completedAbruptly = true;
            try {
        
            	//从队列DelayedWorkQueue获得任务
                while (task != null || (task = getTask()) != null) {
        
                    w.lock();
                    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 {
        
                processWorkerExit(w, completedAbruptly);
            }
        }
        
        private Runnable getTask() {
        
            boolean timedOut = false; // Did the last poll() time out?
    
            for (;;) {
        
                int c = ctl.get();
                int rs = runStateOf(c);
    
                if (rs >= SHUTDOWN && (rs >= STOP || workQueue.isEmpty())) {
        
                    decrementWorkerCount();
                    return null;
                }
    
                int wc = workerCountOf(c);
    
                // Are workers subject to culling?
                boolean timed = allowCoreThreadTimeOut || wc > corePoolSize;
    
                if ((wc > maximumPoolSize || (timed && timedOut))
                    && (wc > 1 || workQueue.isEmpty())) {
        
                    if (compareAndDecrementWorkerCount(c))
                        return null;
                    continue;
                }
    
                try {
        
                	//获得任务
                    Runnable r = timed ?
                        workQueue.poll(keepAliveTime, TimeUnit.NANOSECONDS) :
                        workQueue.take();
                    if (r != null)
                        return r;
                    timedOut = true;
                } catch (InterruptedException retry) {
        
                    timedOut = false;
                }
            }
        }

DelayedWorkQueue是根据触发时间排列的二叉堆，堆顶的是触发时间戳最小的  
进入DelayedWorkQueue的taker方法

public RunnableScheduledFuture<?> take() throws InterruptedException {
        
                final ReentrantLock lock = this.lock;
                //加锁
                lock.lockInterruptibly();
                try {
        
                    for (;;) {
        
                    	//获得二叉堆堆顶的RunnableScheduledFuture
                        RunnableScheduledFuture<?> first = queue[0];
                        //没有任务阻塞
                        if (first == null)
                            available.await();
                        else {
        
                        	//获得任务的触发时间
                            long delay = first.getDelay(NANOSECONDS);
                            //如果时间小于等于0，说明到了触发时间，返回first
                            if (delay <= 0)
                                return finishPoll(first);
                            first = null; // don't retain ref while waiting
                            //如果触发时间没到
                            //如果已经有线程在执行任务，阻塞
                            if (leader != null)
                                available.await();
                            else {
        
                            	//设置这个任务的leader为当前线程
                                Thread thisThread = Thread.currentThread();
                                leader = thisThread;
                                try {
        
                                	//如果触发时间没到，阻塞delay后唤醒
                                    available.awaitNanos(delay);
                                } finally {
        
                                    if (leader == thisThread)
                                        leader = null;
                                }
                            }
                        }
                    }
                } finally {
        
                    if (leader == null && queue[0] != null)
                        available.signal();
                    lock.unlock();
                }
            }

#### 周期性执行 ####

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，比schedule()方法多了一个周期的参数
            ScheduledFutureTask<Void> sft =
                new ScheduledFutureTask<Void>(command,
                                              null,
                                              triggerTime(initialDelay, unit),
                                              //比schedule()方法多了一个周期的参数
                                              //scheduleWithFixedDelay这里是负数，如果是scheduleAtFixedRate（）方法这里是正数
                                              unit.toNanos(-delay));
            RunnableScheduledFuture<Void> t = decorateTask(command, sft);
            sft.outerTask = t;
            delayedExecute(t);
            return t;
        }

ScheduledFutureTask实现周期执行

private class ScheduledFutureTask<V>
                extends FutureTask<V> implements RunnableScheduledFuture<V> {
        
            private final long sequenceNumber;
            //延迟时间
            private long time;
    		//周期执行的间隔
            private final long period;
    
            RunnableScheduledFuture<V> outerTask = this;
    
            int heapIndex;
    //省略代码。。
            ScheduledFutureTask(Runnable r, V result, long ns, long period) {
        
                super(r, result);
                this.time = ns;
                this.period = period;
                this.sequenceNumber = sequencer.getAndIncrement();
            }
    //省略代码
    		// 实现Delayed接口
            public long getDelay(TimeUnit unit) {
        
                return unit.convert(time - now(), NANOSECONDS);
            }
    		// 实现Comparable接口
            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;
            }
            
            //判断是否是周期任务
            public boolean isPeriodic() {
        
                return period != 0;
            }
    		
    		//设置下一次的执行时间
            private void setNextRunTime() {
        
                long p = period;
                if (p > 0)
                    time += p;
                else
                    time = triggerTime(-p);
            }
    
            public boolean cancel(boolean mayInterruptIfRunning) {
        
                boolean cancelled = super.cancel(mayInterruptIfRunning);
                if (cancelled && removeOnCancel && heapIndex >= 0)
                    remove(this);
                return cancelled;
            }
    
            /**
             * Overrides FutureTask version so as to reset/requeue if periodic.
             */
            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);
                }
            }
        }

void reExecutePeriodic(RunnableScheduledFuture<?> task) {
        
            if (canRunInCurrentRunState(true)) {
        
            	//任务重新入队列
                super.getQueue().add(task);
                if (!canRunInCurrentRunState(true) && remove(task))
                    task.cancel(false);
                else
                    ensurePrestart();
            }
        }
        //如果是atFixedRate，period>0，下一次开始执行时间等于上一次开始执行时间+period;
      //如果是withFixedDelay，period < 0，下一次开始执行时间等于为上一次执行的结束时间+(-period)。
    
        long triggerTime(long delay) {
        
        	//now()即为上次任务的结束时间
            return now() +
                ((delay < (Long.MAX_VALUE >> 1)) ? delay : overflowFree(delay));
        }