Spring Boot 3 Scheduler

spring 的 scheduler task 主要需要 spring-context library，因為任何一個 spring boot start 都有包含 spring-context，所以都有支援 scheduler。

spring 提供自動設定類別

• TaskExecutionAutoConfiguration

• TaskSchedulingAutoConfiguration

註冊在 org.springframework.boot.autoconfigure.AutoConfiguration.imports 檔案中

如果 context 裡面沒有任何 Executor instance，就會自動註冊一個預設的 ThreadPoolTaskExecutor，該類別是使用 java.util.concurrent.Executor interface 實作自己的 task executor。同時註冊了一個 TaskExecutorBuilder instance，用來實作自訂的 task executor thread pool。

參數綁定類別是 TaskExecutionProperties

參數是 spring.task.execution.*

---

如果 context 裡面沒有任何一個 instance

• SchedulerConfigurer

  org.springframework.scheduling.annotation

• TaskScheduler

  org.springframework.scheduling

• SchedulerdExecutorService

  java.util.concurrent

就會自動註冊一個 ThreadPoolTaskScheduler instance。

可以註冊一個 TaskSchedulerBuilder instance，參數對應綁定類別是 TaskSchedulingProperties。參數是 spring.task.scheduling.*

---

• thread size < core thread size(coreSize)

  • 產生新的 thread 執行 task

• thread size = core thread size(coreSize) 且 queueCapacity is not full

  • 將 task 放入 queue

• thread size = core thread size(coreSize) 且 queueCapacity is full 且 thread size < max thread size (maxSize)

  • 產生新的 thread 執行 task

• coreSize, queueCapacity, maxSize 都滿了

  • 拒絕執行 task

  • 預設為 AbortPolicy (ref: ExecutorConfigurationSupport)

• thread 空閒時間到達 keepAlive 時

  • 刪除 thread

  • 預設只會刪除 core thread 以外的 threads

  • allowCoreThreadTimeout 可控制是否要刪除 core thread

---

spring task 使用起來比 quartz 簡單

先在 application 加上 @EnableScheduling

@EnableScheduling
@SpringBootApplication
public class TaskApplication {

    public static void main(String[] args) {
        SpringApplication.run(TaskApplication.class, args);
    }

}

• @EnableScheduling

  表示啟用 scheduling，會引用 SchedulingConfiguraiton 自動設定，這是透過註冊 ScheduledAnnotationBeanPostProcessor instance 實現

• @EnableAsync

  啟用 task 非同步運作，這是引用 AsyncConfigurationSelector，根據不同的 AOP 選擇不同的設定類別

這兩個都屬於 spring-context 的 annotation，不需要其他 library

---

SimpleTask

SmpleTask.java

package com.test.task;

import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class SimpleTask {

    @Async
    @Scheduled(cron = "*/3 * * * * *")
    public void printTask() {
        log.info("SimpleTask is working ...");
    }

}

執行時，固定會以 "scheduling-1" 這個 thread 執行

[   scheduling-1] com.test.task.SimpleTask                 : SimpleTask is working ...

沒有用到 ThreadPoolTaskExecutor

---

修改 application，加上 @EnableAsync，才會使用到 ThreadPoolTaskExecutor

@EnableScheduling
@EnableAsync
@SpringBootApplication
public class TaskApplication {

    public static void main(String[] args) {
        SpringApplication.run(TaskApplication.class, args);
    }

}

執行結果，會是 task-1 ~ task-8，然後一直重複。 ThreadPoolTaskExecutor 預設有 8 個 coreThread

[         task-1] com.test.task.SimpleTask                 : SimpleTask is working ...
[         task-2] com.test.task.SimpleTask                 : SimpleTask is working ...

cron

<second> <minute> <hour> <day of month> <month> <day of week>

• second: 0~59

• minute: 0~59

• hour: 0~23

• day of month: 1~31

• month: 1~12 或 JAN ~ DEC

• day of week: 0~7，0 代表週日，或是 MON ~ SUN

example:

cron expression	desc
*/3 * * * * *	每 3s 執行一次
0 0 * * * *	每小時 0分 0 秒執行一次
0 0 9~16 * * *	每天 9 ~ 16 點整點執行一次
0 0 9,16 * * *	每天 9,16 點執行一次
0 0/30 9-16 * * *	每天 9 ~ 16 點，每 30 分執行一次

支援用以下的寫法替代

• @yearly 等同 0 0 0 1 1 *

• @monthly

• @weekly

• @daily

• @hourly

cron expression 不支援年

---

自訂 thread pool

修改系統設定參數

application.yml

spring:
  task:
    execution:
      pool:
        # core thread size
        coreSize: 5
        # max thread size
        maxSize: 10
        # max queue size
        queueCapacity: 50
        # thread keep alive for 10s
        keepAlive: 10s
        #
        allowCoreThreadTimeout: false
    scheduling:
      pool:
        size: 3

拒絕的策略無法透過參數設定修改，只能使用預設的 AbortPolicy

---

自訂 thread pool bean

TaskConfig.java 產生兩個 thread pool: taskExecutor1, taskExecutor2

package com.test.task;

import org.springframework.boot.autoconfigure.task.TaskExecutionProperties;
import org.springframework.boot.context.properties.PropertyMapper;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;

import java.time.Duration;
import java.util.concurrent.ThreadPoolExecutor;

@Configuration
public class TaskConfig {

    @Lazy
    @Bean
    @Primary
    public ThreadPoolTaskExecutor taskExecutor1(TaskExecutionProperties taskExecutionProperties) {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        TaskExecutionProperties.Pool pool = taskExecutionProperties.getPool();
        map.from(pool::getQueueCapacity).to(taskExecutor::setQueueCapacity);
        map.from(pool::getCoreSize).to(taskExecutor::setCorePoolSize);
        map.from(pool::getMaxSize).to(taskExecutor::setMaxPoolSize);
        map.from(pool::getKeepAlive).asInt(Duration::getSeconds).to(taskExecutor::setKeepAliveSeconds);
        map.from(pool::isAllowCoreThreadTimeout).to(taskExecutor::setAllowCoreThreadTimeOut);
        map.from("my-task1-").whenHasText().to(taskExecutor::setThreadNamePrefix);

        // default: AbortPolicy
        taskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy());

        return taskExecutor;
    }

    @Lazy
    @Bean
    public ThreadPoolTaskExecutor taskExecutor2(TaskExecutionProperties taskExecutionProperties) {
        ThreadPoolTaskExecutor taskExecutor = new ThreadPoolTaskExecutor();

        PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
        TaskExecutionProperties.Pool pool = taskExecutionProperties.getPool();
        map.from(10).to(taskExecutor::setQueueCapacity);
        map.from(3).to(taskExecutor::setCorePoolSize);
        map.from(5).to(taskExecutor::setMaxPoolSize);
        map.from(20).to(taskExecutor::setKeepAliveSeconds);
        map.from(true).to(taskExecutor::setAllowCoreThreadTimeOut);
        map.from("my-task2-").whenHasText().to(taskExecutor::setThreadNamePrefix);

        return taskExecutor;
    }
}

修改剛剛的 SimpleTask

@Async("taskExecutor2") 加上 taskExecutor2，指定 thread pool

@Slf4j
@Component
public class SimpleTask {

    @Async("taskExecutor2")
    @Scheduled(cron = "*/3 * * * * *")
    public void printTask() {
        log.info("SimpleTask is working ...");
    }

}

執行結果

[     my-task2-1] com.test.task.SimpleTask                 : SimpleTask is working ...
[     my-task2-2] com.test.task.SimpleTask                 : SimpleTask is working ...
[     my-task2-3] com.test.task.SimpleTask                 : SimpleTask is working ...
[     my-task2-1] com.test.task.SimpleTask                 : SimpleTask is working ...