搞懂Spring任務(wù)執(zhí)行器和調(diào)度器模型
在日常開發(fā)過程中，如果想要處理長時間運(yùn)行的任務(wù)，對于任何應(yīng)用程序開發(fā)而言都不是一件容易的事情。有時候，需要異步執(zhí)行任務(wù)或在特定延遲之后執(zhí)行任務(wù)，這可以通過 Spring 的任務(wù)執(zhí)行和任務(wù)調(diào)度來完成。Spring 框架通過 TaskExecutor 和 TaskScheduler 這兩個接口引入了對異步執(zhí)行和任務(wù)調(diào)度的抽象。讓我們一起來看一下。
Spring 任務(wù)執(zhí)行器
在介紹 Spring 任務(wù)執(zhí)行器之前，要先引出 JDK 中的一個基礎(chǔ)并發(fā)編程組件，即 Executor。所謂的 Executor，本質(zhì)上是在所有內(nèi)部線程任務(wù)執(zhí)行過程上提供了一個抽象層，并管理線程的整個并發(fā)執(zhí)行流。Executor 是執(zhí)行任務(wù)的入口， JDK 為 Executor 提供了以下三個基本接口，即 Executor、ExecutorService 和 ScheduledExecutorService。
圖片
Spring 提供 TaskExecutor 接口作為 Executor 的抽象。TaskExecutor 的實(shí)現(xiàn)類有很多，它們提供了針對異步執(zhí)行過程的各種支持。
圖片
接下來，以最基礎(chǔ)的 SimpleAsyncTaskExecutor為例，來看看在 Spring 應(yīng)用程序中執(zhí)行任務(wù)的具體實(shí)現(xiàn)過程。
TaskExecutor 應(yīng)用方式
SimpleAsyncTaskExecutor 為每個任務(wù)創(chuàng)建一個新線程，并以異步方式運(yùn)行，它實(shí)現(xiàn)了 AsyncTaskExecutor 接口?？梢酝ㄟ^以下方式在 Spring 應(yīng)用程序中注入一個 SimpleAsyncTaskExecutor。
@Bean
AsyncTaskExecutor taskExecutor() {
SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
return taskExecutor;
}然后，可以構(gòu)建一個 AsyncTask，并通過 SimpleAsyncTaskExecutor 來異步執(zhí)行任務(wù)。
public class AsyncTask {
@Autowired
private AsyncTaskExecutor executor;
public void runTasks() throws Exception {
for (int i = 1; i <= 5; i++) {
Runnable task = new SpringTask(" " + i);
executor.execute(task);
}
}
}這里創(chuàng)建了一個自定義的 SpringTask，這是 JDK 中 Runnable 接口的實(shí)現(xiàn)類。
public class SpringTask implements Runnable {
privatestaticfinal Logger LOGGER = Logger.getLogger(Task.class);
private String taskNumber;
public SpringTask(String taskNumber) {
this.taskNumber = taskNumber;
}
@Override
public void run() {
LOGGER.info(Thread.currentThread().getName() + ", Execute Task = " + taskNumber);
taskProcess();
LOGGER.info(Thread.currentThread().getName() + ", End");
}
private void taskProcess() {
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}這里通過讓執(zhí)行線程睡眠 2 秒鐘的方式來模擬任務(wù)的執(zhí)行時間，并把執(zhí)行的過程通過日志的方式打印出來。執(zhí)行這段代碼，會在控制臺中得到如下輸出：
INFO SpringTask:15 - SimpleAsyncTaskExecutor-3, Execute Task = 3
INFO SpringTask:15 - SimpleAsyncTaskExecutor-1, Execute Task = 1
INFO SpringTask:15 - SimpleAsyncTaskExecutor-2, Execute Task = 2
INFO SpringTask:15 - SimpleAsyncTaskExecutor-5, Execute Task = 5
INFO SpringTask:15 - SimpleAsyncTaskExecutor-4, Execute Task = 4
INFO SpringTask:17 - SimpleAsyncTaskExecutor-2, End
INFO SpringTask:17 - SimpleAsyncTaskExecutor-4, End
INFO SpringTask:17 - SimpleAsyncTaskExecutor-3, End
INFO SpringTask:17 - SimpleAsyncTaskExecutor-5, End
INFO SpringTask:17 - SimpleAsyncTaskExecutor-1, End顯然，基于 SimpleAsyncTaskExecutor，任務(wù)與線程之間應(yīng)該是一對一的執(zhí)行關(guān)系。
TaskExecutor 運(yùn)行原理
介紹完 Spring 任務(wù)執(zhí)行器的使用方式之后，來進(jìn)一步分析它的實(shí)現(xiàn)原理。同樣，還是以前面介紹的 SimpleAsyncTaskExecutor 為例展開討論。SimpleAsyncTaskExecutor 的類層結(jié)構(gòu)如圖所示：
圖片
這里的 TaskExecutor 和 AsyncTaskExecutor 都比較好理解，直接看它們的接口定義，如下所示：
public interface TaskExecutor extends Executor {
@Override
void execute(Runnable task);
}
public interface AsyncTaskExecutor extends TaskExecutor {
void execute(Runnable task, long startTimeout);
Future<?> submit(Runnable task);
<T> Future<T> submit(Callable<T> task);
}AsyncListenableTaskExecutor 又?jǐn)U展了 AsyncTaskExecutor，添加了可以返回 ListenableFuture 的方法，ListenableFuture 是 JDK 中 Future 接口的子接口，可用于在任務(wù)提交后添加回調(diào)。
接下來，來看 AsyncListenableTaskExecutor 中的 execute 方法，如下所示：
@Override
public void execute(Runnable task, long startTimeout) {
Assert.notNull(task, "Runnable must not be null");
//使用包裝器包裝任務(wù)
Runnable taskToUse = (this.taskDecorator != null ? this.taskDecorator.decorate(task) : task);
//啟用限流器來執(zhí)行任務(wù)
if (isThrottleActive() && startTimeout > TIMEOUT_IMMEDIATE) {
this.concurrencyThrottle.beforeAccess();
doExecute(new ConcurrencyThrottlingRunnable(taskToUse));
}
//不啟用限流器來執(zhí)行任務(wù)
else {
doExecute(taskToUse);
}
}這里引出了限流器的概念，限流器的作用是線程執(zhí)行的并發(fā)度達(dá)到閾值則會讓后續(xù)的線程處于阻塞等待。這是 Spring TaskExecutor 設(shè)計上的一個亮點(diǎn)，基本思想如圖所示：
圖片
結(jié)合圖示，不難看出限流器是在線程執(zhí)行之前進(jìn)行并發(fā)限制的判斷，如果需要限流就阻塞線程。而如果任務(wù)執(zhí)行完成后，那就喚醒正在等待的線程繼續(xù)執(zhí)行任務(wù)。
而真正執(zhí)行任務(wù)的 doExecute() 方法比較簡單，單獨(dú)從線程工廠 ThreadFactory 獲取線程，或者直接創(chuàng)建一個新的線程進(jìn)行執(zhí)行即可，如下所示：
protected void doExecute(Runnable task) {
Thread thread = (this.threadFactory != null ? this.threadFactory.newThread(task) : createThread(task));
thread.start();
}Spring 任務(wù)調(diào)度器
介紹完 Spring 任務(wù)執(zhí)行器，接下來討論 Spring 任務(wù)調(diào)度器。有時，我們需要以固定的時間間隔執(zhí)行任務(wù)，就可以通過任務(wù)調(diào)度器來實(shí)現(xiàn)?；?Spring，將看到如何使用一些注解來對任務(wù)進(jìn)行調(diào)度。
TaskScheduler 應(yīng)用方式
在 Spring 中，可以借助@EnableScheduling 注解來啟用任務(wù)調(diào)度。
@Configuration
@EnableScheduling
public class SpringSchedulingExample一旦啟用了任務(wù)調(diào)度，Spring 將自動注冊一個內(nèi)部 BeanPostProcessor，它將在 Spring 管理的 Bean 上找到添加了@Scheduled 注解的方法。@Scheduled 注解的使用方法如下所示：
@Scheduled(fixedDelay = 2000)
public void scheduledTask() {
LOGGER.info("Execute task " + new Date());
}在這里，使用@Scheduled 注解為 scheduledTask() 設(shè)置了調(diào)度任務(wù)，即通過 fixedDelay 屬性每兩秒執(zhí)行一次該方法。執(zhí)行該方法，可以在控制臺上看到如下信息：
INFO SpringSchedulingExample:17 - Execute task Sat May 01 20:06:11 CST 2021
INFO SpringSchedulingExample:17 - Execute task Sat May 01 20:06:13 CST 2021
INFO SpringSchedulingExample:17 - Execute task Sat May 01 20:06:15 CST 2021
INFO SpringSchedulingExample:17 - Execute task Sat May 01 20:06:17 CST 2021
INFO SpringSchedulingExample:17 - Execute task Sat May 01 20:06:19 CST 2021當(dāng)然，還可以使用其他調(diào)度屬性，@Scheduled 注解的完整定義如下所示：
@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Repeatable(Schedules.class)
public @interface Scheduled {
//cron 表達(dá)式
String cron() default "";
//由 cron 表達(dá)式進(jìn)行解析的時間區(qū)域
String zone() default "";
//固定延遲時間
long fixedDelay() default -1;
//字符串形式的固定延遲時間
String fixedDelayString() default "";
//固定周期
long fixedRate() default -1;
//字符串形式的固定周期
String fixedRateString() default "";
//初始延遲時間
long initialDelay() default -1;
//字符串形式的初始延遲時間
String initialDelayString() default "";
}這些屬性都很簡單，唯一需要說明一下的是 fixedRate。舉個例子，如果在某個方法上設(shè)置 fiexdRate=3000，而執(zhí)行該方法所花的時間是一秒，那么兩秒后就會再次執(zhí)行該方法。
TaskScheduler 運(yùn)行原理
在 Spring 的 TaskScheduler 出現(xiàn)之前，可以使用 JDK 中的 Timer 或第三方的 Quartz 組件類實(shí)現(xiàn)調(diào)度功能。而 TaskScheduler 的核心優(yōu)勢就是為開發(fā)人員提供了一種抽象，使得執(zhí)行定時任務(wù)的代碼不需要指定特定的定時框架。TaskScheduler 接口的定義如下所示：
public interface TaskScheduler {
ScheduledFuture<?> schedule(Runnable task, Trigger trigger);
ScheduledFuture<?> schedule(Runnable task, Date startTime);
ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period);
ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period);
ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, Date startTime, long delay);
ScheduledFuture<?> scheduleWithFixedDelay(Runnable task, long delay);
}TaskScheduler 中各個方法的含義可以結(jié)合@Scheduled 注解的說明進(jìn)行理解，而這些方法都返回了一個 JDK 中的 ScheduledFuture 對象，該對象是對 Future 的擴(kuò)展。
在 Spring 中，TaskScheduler 接口的代表性實(shí)現(xiàn)就是 ThreadPoolTaskScheduler，其類層結(jié)構(gòu)如圖所示：
圖片
翻閱 ThreadPoolTaskScheduler 類的源代碼，發(fā)現(xiàn)在該類中存在如下所示的一個變量定義。
private ScheduledExecutorService scheduledExecutor;在前面介紹 JDK Executor 時，已經(jīng)引出了 ScheduledExecutorService，它為開發(fā)人員提供了各種調(diào)度方法。所以，看到這里，不難理解，ThreadPoolTaskScheduler 實(shí)際上就是將各種調(diào)度操作委托給了這個 ScheduledExecutorService。通過如下所示的 schedule 方法實(shí)現(xiàn)過程印證了這一點(diǎn)。
@Override
public ScheduledFuture<?> schedule(Runnable task, Date startTime) {
ScheduledExecutorService executor = getScheduledExecutor();
long initialDelay = startTime.getTime() - System.currentTimeMillis();
try {
return executor.schedule(errorHandlingTask(task, false), initialDelay, TimeUnit.MILLISECONDS);
}
catch (RejectedExecutionException ex) {
throw new TaskRejectedException("Executor [" + executor + "] did not accept task: " + task, ex);
}
}可以看到，這里首先獲取一個 ScheduledExecutorService，然后通過它的 schedule 方法完成調(diào)度。而 ScheduledExecutorService 的創(chuàng)建過程也很簡單，如下所示：
protected ScheduledExecutorService createExecutor(
int poolSize, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
return new ScheduledThreadPoolExecutor(poolSize, threadFactory, rejectedExecutionHandler);
}這樣，就把 JDK 中的 ScheduledExecutorService 和 Spring 中的 TaskScheduler 關(guān)聯(lián)了起來，從而完成了對任務(wù)調(diào)度過程的剖析。
總結(jié)
針對并發(fā)編程，我們可以使用 JDK 所提供的 Thread 類和 Runnable 接口來創(chuàng)建和管理多線程。但由于這些技術(shù)組件過于底層，所以在日常開發(fā)過程中，我一般不推薦你使用它們來創(chuàng)建多線程應(yīng)用程序。
而 JDK 并發(fā)包中的 Executor 等組件雖然經(jīng)過高度抽象，為開發(fā)人員提供了高層次的 API，但由于并發(fā)編程涉及到多線程之間的協(xié)作和交互，合理使用這些組件對開發(fā)人員的要求也很高。
Spring 框架充分考慮到了這些問題，并結(jié)合常見的應(yīng)用場景提供了任務(wù)執(zhí)行器和任務(wù)調(diào)度器組件。在今天的內(nèi)容中，我們從應(yīng)用方法和 運(yùn)行原理這兩個維度對這兩款技術(shù)組件進(jìn)行了詳細(xì)地分析，幫助你在開發(fā)過程中能夠更好地實(shí)現(xiàn)各種并發(fā)編程需求。
圖片
