1.調(diào)度服務

1.1.JDK之ScheduledExecutorService

講到調(diào)度任務，我們腦海里馬上會想到ScheduledExecutorService。

ScheduledExecutorService是 Java java.util.concurrent 包中的一個接口，它繼承自 ExecutorService 接口。它主要用于在給定的延遲后運行任務，或者定期地執(zhí)行任務。這個接口提供了幾種安排任務執(zhí)行的方法，包括單次執(zhí)行、定期執(zhí)行和周期性執(zhí)行。

以下是 ScheduledExecutorService 提供的一些關鍵方法：

• schedule(Callable<V> callable, long delay, TimeUnit unit)： 安排所提交的 Callable 任務在指定的延遲后運行，返回一個 Future，代表任務的結果。
• schedule(Runnable command, long delay, TimeUnit unit)： 安排所提交的 Runnable 任務在指定的延遲后運行。
• scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit)： 安排所提交的 Runnable 任務在指定的初始延遲后首次啟動，并且隨后按指定的周期重復執(zhí)行。
• scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit)： 安排所提交的 Runnable 任務在指定的初始延遲后首次啟動，并且隨后在每次執(zhí)行結束和下次執(zhí)行開始之間都存在指定的延遲。

然而，如果采用了sprintboot，我們也可以直接采用springboot提供的調(diào)試線程池。這其中有一個最大的優(yōu)勢在于，可以利用spring的cron表達式，采用方法注解的方式，不用引入quartz第三方工具。

1.2.springboot使用調(diào)度線程池

我們嘗試從源代碼的角度，來看看springboot提供的調(diào)度線程池是怎么創(chuàng)建的。

首先，啟用調(diào)度配置，啟動類加上@EnableScheduling

springboot的自動配置類一般以AutoConfiguration作為后綴，采用模糊搜索ScheduleAutoConfiguration可以找到目標TaskSchedulingAutoConfiguration。

從截圖我們可以看出，我們只需在application.yml加入以下的配置就可以啟動了

spring:
  task:
    ## 定時任務(業(yè)務上定時任務量不多，2個足矣)
    scheduling:
      ## 線程池核心線程數(shù)量
      pool:
        size: 2
      ## 線程名稱前線
      threadNamePrefix: common-scheduling-

spring對線程池進行二次封閉，最終調(diào)用的還是jdk的ThreadPoolExecutor類，我們在該類的構造函數(shù)打個斷點，可以看到，pool.size參數(shù)已經(jīng)被傳參：

1.3.ThreadPoolExecutor核心參數(shù)與執(zhí)行流程

這里有必要先介紹下 ThreadPoolExecutor類的幾個函數(shù)參數(shù)及基本運行機制

構造函數(shù)參數(shù)：

• 核心線程數(shù)（Core Pool Size）： 線程池中始終保持的線程數(shù)量，即使它們處于空閑狀態(tài)。如果任務數(shù)量少于核心線程數(shù)，線程池會創(chuàng)建新的線程來處理任務，而不會立即回收這些線程。
• 最大線程數(shù)（Maximum Pool Size）： 線程池中允許的最大線程數(shù)量。如果任務數(shù)量超過了核心線程數(shù)但小于最大線程數(shù)，且工作隊列已滿，線程池會創(chuàng)建新的線程來處理任務，直到達到最大線程數(shù)。
• 工作隊列（Work Queue）： 用于存放待執(zhí)行任務的阻塞隊列。當所有核心線程都在忙碌時，新的任務會被放入工作隊列中等待執(zhí)行。
• 線程工廠（Thread Factory）： 用于創(chuàng)建新線程的工廠。它提供了一種方式來定制線程的創(chuàng)建過程，例如設置線程的名稱、優(yōu)先級、是否為守護線程等。
• 拒絕策略（Rejected Execution Handler）： 當任務無法被線程池及時處理時（即當線程池已滿，且工作隊列已滿），線程池會采用拒絕策略來處理新提交的任務。常見的拒絕策略包括：
• AbortPolicy：拋出 RejectedExecutionException。
• CallerRunsPolicy：由調(diào)用者線程運行該任務。
• DiscardPolicy：靜默丟棄任務。
• DiscardOldestPolicy：丟棄隊列中最老的任務，然后嘗試再次提交當前任務。
• 保持活動時間（Keep-Alive Time）： 非核心線程空閑時在終止前等待新任務的最長時間。如果線程池允許核心線程空閑，這個參數(shù)也適用于核心線程。
• 時間單位（Time Unit）： 與保持活動時間配合使用的時間單位，例如 TimeUnit.SECONDS。

執(zhí)行流程：

• 如果線程池中的線程數(shù)量少于核心線程數(shù)，即使有空閑線程，線程池也會優(yōu)先創(chuàng)建新線程來執(zhí)行新的任務。
• 如果線程池中的線程數(shù)量達到核心線程數(shù)，新的任務會被放入工作隊列等待執(zhí)行。
• 如果工作隊列已滿且線程數(shù)量少于最大線程數(shù)，線程池會創(chuàng)建新的非核心線程來執(zhí)行任務。
• 如果工作隊列已滿且線程數(shù)量達到最大線程數(shù)，新的任務會被拒絕，線程池會采用拒絕策略來處理。

1.4.自定義線程池

由于spingboot對調(diào)度任務線程池的參數(shù)支持有限，如果想定制自己的參數(shù)，可以注入自己的調(diào)度線程池，從代碼可看出：

@Configuration
public class ThreadPoolConfig {

    @Bean
    public ScheduledExecutorService scheduledExecutorService() {
        return new ScheduledThreadPoolExecutor(2,
                new NamedThreadFactory("common-schedule"),
                new ThreadPoolExecutor.CallerRunsPolicy()) {
            @Override
            protected void afterExecute(Runnable r, Throwable t) {
                super.afterExecute(r, t);
            }
        };
    }

}

1.5.spring cron注解

Cron 表達式是一種用于描述定時任務觸發(fā)時間的字符串表達式。它由多個時間字段組成，每個字段代表定時任務在特定時間單位上的觸發(fā)條件。

1.5.1.cron表達式語法格式

秒 分 時 日 月 星期 年份

其中，每個時間字段都有對應的取值范圍和特殊符號。下面是每個時間字段的詳細說明：

1、秒（Seconds）：取值范圍為 0~59。例如，`0/5` 表示從0秒開始，每隔 5 秒觸發(fā)一次，`*` 表示每秒都觸發(fā)。

2、分鐘（Minutes）：取值范圍為 0~59。例如，`0/5` 表示從0分鐘開始，每隔 5 分鐘觸發(fā)一次，`*` 表示每分鐘都觸發(fā)。

3、小時（Hours）：取值范圍為 0~23。例如，`0/2` 表示從0小時開始，每隔 2 小時觸發(fā)一次，`*` 表示每小時都觸發(fā)。

4、日期（Day of Month）：取值范圍為 1~31。例如，`1,15` 表示每月的 1 日和 15 日觸發(fā)，`*` 表示每天都觸發(fā)。

5、月份（Month）：取值范圍為 1~12，也可以使用英文縮寫 JAN、FEB、MAR 等。例如，`1,6` 表示一月和六月觸發(fā)，`*` 表示每個月都觸發(fā)。

6、 星期（Day of Week）：取值范圍為 1~7，1 表示星期日，2 表示星期一，以此類推，也可以使用英文縮寫 SUN、MON、TUE 等。例如，`2-6` 表示星期一到星期五觸發(fā)，`*` 表示每個星期都觸發(fā)。

7、年份（Year）：可選字段，表示觸發(fā)條件的年份。例如，`2023` 表示在 2023 年觸發(fā)，`*` 表示每年都觸發(fā)。

除了取值范圍，Cron 表達式還支持一些特殊符號，用于指定特定的觸發(fā)條件，例如：

• - 星號（*）：代表所有可能的取值，表示不限制該時間字段的取值范圍。
• - 問號（?）：僅在日期和星期字段中使用，表示不指定具體的取值，可以任意匹配。
• - 斜線（/）：表示間隔觸發(fā)，例如在分鐘字段中，"*/5" 表示每隔 5 分鐘觸發(fā)一次。
• - 逗號（,）：用于指定多個取值，例如在小時字段中，"1,3,5" 表示在第 1、3、5 小時觸發(fā)。
• - 減號（-）：用于指定一個范圍，例如在月份字段中，"3-6" 表示三月到六月觸發(fā)。
• L : 表示最后，只能出現(xiàn)在星期和每月第幾天域，如果在星期域使用1L，意味著在最后的一個星期日觸發(fā)。
• W : 表示有效工作日(周一到周五),只能出現(xiàn)在每月第幾日域，系統(tǒng)將在離指定日期的最近的有效工作日觸發(fā)事件。注意一點，W的最近尋找不會跨過月份
• LW : 這兩個字符可以連用，表示在某個月最后一個工作日，即最后一個星期五。

# : 用于確定每個月第幾個星期幾，只能出現(xiàn)在每月第幾天域。例如在1#3，表示某月的第三個星期日。

1.5.2.cron表達式示例

2.異步任務

2.1.springboot配置

在軟件開發(fā)中，有些任務比較耗時但又無需馬上獲得結果。一般地，這些任務我們可以采用獨立線程池異步執(zhí)行。如果程序基于springboot環(huán)境，我們有現(xiàn)成的工具可以使用。

首先，我們需要在程序啟動入口類增加@EnableAsync。

借著，我們嘗試從源代碼的角度，來看看springboot提供的異步線程池是怎么創(chuàng)建的。

從springboot的命名風格可知，通過模糊搜索TaskAutoConfiguration，可以找到TaskExecutionAutoConfiguration，如下：

從源代碼可知，只需在application.yml配置如下參數(shù)即可：

spring:    
    execution:
      pool:
        coreSize: 4
        queueCapacity: 64
        maxSize: 8
        ## 禁止空閑線程關閉，保證最少有core個存活線程
        allowCoreThreadTimeout: false
        keepAlive: 300s
      threadNamePrefix: common-async_task-

這些參數(shù)跟jdk的ThreadPoolExecutor類的構造參數(shù)非常相似，這里不作解析 。

2.2.使用異步任務

使用方法，只要在目標方法的簽名加上@Async

然而，執(zhí)行結果卻出乎意外（在main主線程上執(zhí)行，沒有異步執(zhí)行）

熟悉springaop機制的同學，馬上知道這是因為異步任務底層是基于動態(tài)代理機制實現(xiàn)的。Spring AOP 代理只有在通過 Spring 容器獲取 Bean 時才會創(chuàng)建。

當在同一個類內(nèi)部調(diào)用一個方法時，調(diào)用的是原始對象，而不是代理對象。因此，內(nèi)部調(diào)用不會經(jīng)過 Spring AOP 代理，也就無法觸發(fā)異步執(zhí)行。解決這個問題最簡單的方法是用一個新的類來管理異步執(zhí)行方法。

問題解決

2.3.模擬系統(tǒng)繁忙進行性能測試

我們嘗試模擬一些極端情況，系統(tǒng)處理不過來的情況。修改上面的配置，改為

pool:
  coreSize: 1
  queueCapacity: 1
  maxSize: 8

同時，異步執(zhí)行增加時間延遲來模擬耗時任務

@Component
@Slf4j
class AsyncTaskHandler {

    @Async
    public void busyTask1() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignored) {
        }
        log.info("----------busyTask1---------");
    }

    @Async
    public void busyTask2() {
        try {
            Thread.sleep(5000);
        } catch (InterruptedException ignored) {
        }
        log.info("----------busyTask2---------");
    }

}

執(zhí)行結果會出現(xiàn)報錯（線程數(shù)已達到最大數(shù)量，且任務隊列已滿，無法添加新任務）

調(diào)度任務執(zhí)行頻率是可預見的，有多少個任務，執(zhí)行頻率，開發(fā)可知，核心數(shù)量，最大數(shù)量，隊列容量比較好設定。而異步執(zhí)行頻率很大程度是由系統(tǒng)的使用者（用戶）決定的，因此這些參數(shù)需要根據(jù)流量動態(tài)修改。

2.4.異步線程池拒絕策略

如果不想把 queueCapacity和maxSize都設置成很大的話，我們可以考慮修改下線程池的拒絕策略。最妥當?shù)姆绞绞?，既然異步不了，那?quot;熔斷"成同步，直接在調(diào)用者所在的業(yè)務線程執(zhí)行。然而，springboot沒有提供相應的配置項。為此，我們只能關閉springboot的自動配置了。

從代碼可看出，只要提供一個Executor實例，并且名字叫taskExecutor即可。

話不多說，上代碼

@Configuration
public class ThreadPoolConfig {

    private final int core = Runtime.getRuntime().availableProcessors();

    /**
     * springboot 自動注入的異步執(zhí)行線程池，拒絕策略為丟棄，難以配置maxPoolSize參數(shù)
     * @see TaskExecutionAutoConfiguration#taskExecutorBuilder
     */
    @Bean(name = "taskExecutor")
    public ThreadPoolTaskExecutor threadPoolTaskExecutor(TaskExecutionProperties properties) {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(properties.getPool().getCoreSize());
        executor.setThreadNamePrefix(properties.getThreadNamePrefix());
        executor.setMaxPoolSize(properties.getPool().getMaxSize());
        executor.setQueueCapacity(properties.getPool().getQueueCapacity());
        executor.setKeepAliveSeconds((int) properties.getPool().getKeepAlive().toSeconds());
        executor.setAllowCoreThreadTimeOut(properties.getPool().isAllowCoreThreadTimeout());
        // 超過隊列容量，則在業(yè)務線程上執(zhí)行
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        return executor;
    }

}

重新運行程序，可以看到，當任務超過線程池負載的時候，多余的線程會在調(diào)用線程上執(zhí)行，變?yōu)橥酱a

個人認為：使用springboot創(chuàng)建的線程池，代碼也只是稍微簡化一點點。

采用原生線程池，每個業(yè)務代碼必須實現(xiàn)Runnable接口，而使用springboot的異步線程池，只需以方法注解的形式即可，底層aop會生成對應的代理方法。但要確保避免內(nèi)部方法調(diào)用導致異步邏輯失效。

總結

以上為個人經(jīng)驗，希望能給大家一個參考，也希望大家多多支持腳本之家。

最新評論