摘要：今天給大家介紹下周期性線程池的使用和重點源碼剖析。用來處理延時任務(wù)或定時任務(wù)定時線程池類的類結(jié)構(gòu)圖接收類型的任務(wù)，是線程池調(diào)度任務(wù)的最小單位。周期性線程池任務(wù)的提交方式周期性有三種提交的方式。

之前學(xué)習(xí)ThreadPool的使用以及源碼剖析，并且從面試的角度去介紹知識點的解答。今天給大家介紹下周期性線程池的使用和重點源碼剖析。

ScheduledThreadPoolExecutor：用來處理延時任務(wù)或定時任務(wù)
定時線程池類的類結(jié)構(gòu)圖

ScheduledThreadPoolExecutor接收ScheduleFutureTask類型的任務(wù)，是線程池調(diào)度任務(wù)的最小單位。
它采用DelayQueue存儲等待的任務(wù)：
1、DelayQueue內(nèi)部封裝成一個PriorityQueue，它會根據(jù)time的先后時間順序，如果time相同則根絕sequenceNumber排序；
2、DelayQueue是無界隊列；

接收的參數(shù):

private final long sequenceNumber;//任務(wù)的序號
private long time;//任務(wù)開始的時間
private final long period;//任務(wù)執(zhí)行的時間間隔

工作線程的的執(zhí)行過程：
工作線程會從DelayQueue取出已經(jīng)到期的任務(wù)去執(zhí)行；
執(zhí)行結(jié)束后重新設(shè)置任務(wù)的到期時間，再次放回DelayQueue；

ScheduledThreadPoolExecutor會把待執(zhí)行的任務(wù)放到工作隊列DelayQueue中，DelayQueue封裝了一個PriorityQueue，PriorityQueue會對隊列中的ScheduledFutureTask進(jìn)行排序，具體的排序算法實現(xiàn)如下：

public int compareTo(Delayed other) {
    if (other == this) // compare zero if same object
        return 0;
    if (other instanceof ScheduledFutureTask) {
        ScheduledFutureTask x = (ScheduledFutureTask)other;
        //首先按照time排序，time小的排到前面，time大的排到后面
        long diff = time - x.time;
        if (diff < 0)
            return -1;
        else if (diff > 0)
            return 1;
        //time相同，按照sequenceNumber排序;
        //sequenceNumber小的排在前面,sequenceNumber大的排在后面 
        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;
}

接下來看看ScheduledFutureTask的run方法實現(xiàn)， run方法是調(diào)度task的核心，task的執(zhí)行實際上是run方法的執(zhí)行。

public void run() {
    //是否是周期性的
    boolean periodic = isPeriodic();
    //線程池是shundown狀態(tài)不支持處理新任務(wù)，直接取消任務(wù)
    if (!canRunInCurrentRunState(periodic))
        cancel(false);
    //如果不需要執(zhí)行執(zhí)行周期性任務(wù)，直接執(zhí)行run方法結(jié)束
    else if (!periodic)
        ScheduledFutureTask.super.run();
    //如果需要周期性執(zhí)行，則在執(zhí)行任務(wù)完成后，設(shè)置下一次執(zhí)行時間
    else if (ScheduledFutureTask.super.runAndReset()) {
        //設(shè)置下一次執(zhí)行該任務(wù)的時間
        setNextRunTime();
        //重復(fù)執(zhí)行該任務(wù)
        reExecutePeriodic(outerTask);
    }
}

run方法的執(zhí)行步驟：

1、如果線程池是shundown狀態(tài)不支持處理新任務(wù)，直接取消任務(wù)，否則步驟2；

2、如果不是周期性任務(wù)，直接調(diào)用ScheduledFutureTask的run方法執(zhí)行，會設(shè)置執(zhí)行結(jié)果，然后直接返回，否則步驟3；

3、如果是周期性任務(wù)，調(diào)用ScheduledFutureTask的runAndset方法執(zhí)行，不會設(shè)置執(zhí)行結(jié)果，然后直接返回，否則執(zhí)行步驟4和步驟5；

4、計算下一次執(zhí)行該任務(wù)的時間；

5、重復(fù)執(zhí)行該任務(wù)；

接下來看下reExecutePeriodic方法的執(zhí)行步驟：

void reExecutePeriodic(RunnableScheduledFuture task) {
    if (canRunInCurrentRunState(true)) {
        super.getQueue().add(task);
        if (!canRunInCurrentRunState(true) && remove(task))
            task.cancel(false);
        else
            ensurePrestart();
    }
}

由于已經(jīng)執(zhí)行過一次周期性任務(wù)，所以不會reject當(dāng)前任務(wù)，同時傳入的任務(wù)一定是周期性任務(wù)。

周期性有三種提交的方式：schedule、sceduleAtFixedRate、schedlueWithFixedDelay。下面從使用和源碼兩個方面進(jìn)行說明，首先是如果提交任務(wù)：

pool.schedule(new Runnable() {
    @Override
    public void run() {
        System.out.println("延遲執(zhí)行");
    }
},1, TimeUnit.SECONDS);

/**
 * 這個執(zhí)行周期是固定，不管任務(wù)執(zhí)行多長時間，每過3秒中就會產(chǎn)生一個新的任務(wù)
 */
pool.scheduleAtFixedRate(new Runnable() {
    @Override
    public void run() {
        //這個業(yè)務(wù)邏輯需要很長的時間，超過了3秒
        System.out.println("重復(fù)執(zhí)行");
    }
},1,3,TimeUnit.SECONDS);

pool.shutdown();

/**
 * 假如run方法30min后執(zhí)行完成，然后間隔3秒，再周期性執(zhí)行下一個任務(wù)
 */
pool.scheduleWithFixedDelay(new Runnable() {
    @Override
    public void run() {
        //30min
        System.out.println("重復(fù)執(zhí)行");
    }
},1,3,TimeUnit.SECONDS);

知道了如何提交周期性任務(wù)，接下來源碼是如何執(zhí)行的，首先是schedule方法，該方法是指任務(wù)在指定延遲時間到達(dá)后觸發(fā)，只會執(zhí)行一次。

public ScheduledFuture schedule(Runnable command,
                                   long delay,
                                   TimeUnit unit) {
    if (command == null || unit == null)
        throw new NullPointerException();
    //把任務(wù)封裝成ScheduledFutureTask，之后調(diào)用decorateTask進(jìn)行包裝；
    //decorateTask方法是空方法，留給用戶去實現(xiàn)的；
    RunnableScheduledFuture t = decorateTask(command,
        new ScheduledFutureTask(command, null,
                                      triggerTime(delay, unit)));
    //包裝好任務(wù)之后，進(jìn)行任務(wù)的提交                                  
    delayedExecute(t);
    return t;
}

任務(wù)提交方法：

private void delayedExecute(RunnableScheduledFuture task) {
    //如果線程池不是RUNNING狀態(tài)，則使用拒絕策略把提交任務(wù)拒絕掉
    if (isShutdown())
        reject(task);
    else {
        //與ThreadPoolExecutor不同，這里直接把任務(wù)加入延遲隊列
        super.getQueue().add(task);
        //如果當(dāng)前狀態(tài)無法執(zhí)行任務(wù)，則取消
        if (isShutdown() &&
            !canRunInCurrentRunState(task.isPeriodic()) &&
            remove(task))
            task.cancel(false);
        else
        //和ThreadPoolExecutor不一樣，corePoolSize沒有達(dá)到會增加Worker;
        //增加Worker，確保提交的任務(wù)能夠被執(zhí)行
            ensurePrestart();
    }
}

還沒關(guān)注我的公眾號？

掃文末二維碼關(guān)注公眾號【小強(qiáng)的進(jìn)階之路】可領(lǐng)取如下:

學(xué)習(xí)資料： 1T視頻教程：涵蓋Javaweb前后端教學(xué)視頻、機(jī)器學(xué)習(xí)/人工智能教學(xué)視頻、Linux系統(tǒng)教程視頻、雅思考試視頻教程；

100多本書：包含C/C++、Java、Python三門編程語言的經(jīng)典必看圖書、LeetCode題解大全；

軟件工具：幾乎包括你在編程道路上的可能會用到的大部分軟件；

項目源碼：20個JavaWeb項目源碼。