復(fù)盤SpringBoot中定時任務(wù)和異步線程池

復(fù)盤SpringBoot中定時任務(wù)和異步線程池項目中最近使用了多個定時任務(wù)處理業(yè)務(wù)需求，于是在實現(xiàn)業(yè)務(wù)邏輯過程中，產(chǎn)生了上圖一些思考和疑問，現(xiàn)在利用空余時間進行一次復(fù)盤。項目搭建項目搭建環(huán)境：JDK1.8+SpringBoot主啟動類：加上@EnableScheduling新建定時任務(wù)配置類：ScheduledTask；定義兩個定時任務(wù)，簡單打印一下線程名字和時間戳源碼如下：@Component

public

class

ScheduledTask

{

@Scheduled(cron

=

"0/1

*

*

*

*

?")

public

void

scheduledTask1()

throws

InterruptedException

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask1

"

+

System.currentTimeMillis());

}

@Scheduled(cron

=

"0/1

*

*

*

*

?")

public

void

scheduledTask2()

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask2

"

+

System.currentTimeMillis());

}

}一、多任務(wù)串行執(zhí)行1.相同定時任務(wù)先解決多任務(wù)定時相同時間，是否存在優(yōu)先級執(zhí)行順序，執(zhí)行上面的代碼，打印日志如下圖：從控制臺日志發(fā)現(xiàn)，兩個定時任務(wù)并沒有存在一定的執(zhí)行順序，存在亂序現(xiàn)象。故：串行定時任務(wù)，沒有明顯的優(yōu)先級關(guān)系。2.一個定時任務(wù)阻塞為了實現(xiàn)此場景的條件，將定時任務(wù)1中添加死循環(huán)邏輯。源碼改動如下：

@Scheduled(cron

=

"0/1

*

*

*

*

?")

public

void

scheduledTask1()

throws

InterruptedException

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask1

"

+

System.currentTimeMillis());

while

(true)

{

Thread.sleep(5000);

}

}從控制臺可以得出：多個定時任務(wù)時串行執(zhí)行的,如果一個任務(wù)出現(xiàn)阻塞，其他的任務(wù)都會受到影響。二、多任務(wù)并行執(zhí)行如果要實現(xiàn)并行執(zhí)行，啟動類需要在上面的基礎(chǔ)上新增注解@EnableAsync。任務(wù)方法上新增@Async注解。源碼如下：@Component

public

class

ScheduledTask

{

@Scheduled(cron

=

"0/1

*

*

*

*

?")

@Async

public

void

scheduledTask1()

throws

InterruptedException

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask1

"

+

System.currentTimeMillis());

}

@Scheduled(cron

=

"0/1

*

*

*

*

?")

@Async

public

void

scheduledTask2()

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask2

"

+

System.currentTimeMillis());

}

}執(zhí)行結(jié)果如下圖：從控制臺中打印的線程名發(fā)現(xiàn)：每次執(zhí)行任務(wù)時，都是創(chuàng)建新的線程執(zhí)行，使用默認線程池SimpleAsyncTaskExecutor。默認情況下異步調(diào)用使用的線程池是SimpleAsyncTaskExecutor，該線程池是不被推薦，因為該線程池的線程不重用，每次調(diào)用都會創(chuàng)建一個新的線程。所以需要我們自定義線程池。自定義線程池1.自定義局部線程池局部線程池實際上就是指異步方法上需要指定使用該線程池，否則將使用默認線程池。搜索公眾號后端架構(gòu)師后臺回復(fù)“架構(gòu)整潔”，獲取一份驚喜禮包。配置異步線程池源碼如下：@Component

public

class

AsyncTaskExecutorConfig

{

/**

*

重寫AsyncTaskExecutor對象，實現(xiàn)全局異步線程，即@Async注解需指定線程池

*/

@Bean(value

=

"asyncTaskExecutor")

public

AsyncTaskExecutor

asyncTaskExecutor()

{

ThreadPoolTaskExecutor

executor

=

new

ThreadPoolTaskExecutor();

executor.setThreadNamePrefix("defineAsyncTask-");

executor.setMaxPoolSize(10);

executor.setCorePoolSize(3);

executor.setQueueCapacity(100);

/*

線程池對拒絕任務(wù)的處理策略(rejectionpolicy)：

當(dāng)線程池已經(jīng)達到最大線程數(shù)量，沒有空閑線程時，新任務(wù)該如何處理

可選策略：

CallerRunsPolicy:當(dāng)線程池沒有能力處理時直接在執(zhí)行方法的調(diào)用線程中運行被拒絕的任務(wù)

如果執(zhí)行程序已經(jīng)關(guān)閉，將丟棄該任務(wù).

AbortPolicy:處理程序遭到拒絕時將拋出

RejectedExecutionException

*/

executor.setRejectedExecutionHandler(new

ThreadPoolExecutor.CallerRunsPolicy());

//等待所有任務(wù)調(diào)度完成在關(guān)閉線程池，保證所有的任務(wù)被正確處理

executor.setWaitForTasksToCompleteOnShutdown(true);

//線程池關(guān)閉時等待其他任務(wù)的時間，不能無限等待，確保應(yīng)用最后能被關(guān)閉。而不是無限期阻塞

executor.setAwaitTerminationSeconds(60);

//線程池初始化

executor.initialize();

return

executor;

}

}定時任務(wù)源碼修改如下：@Component

public

class

ScheduledTask

{

@Scheduled(cron

=

"0/1

*

*

*

*

?")

//指定自定義線程池

@Async("asyncTaskExecutor")

public

void

scheduledTask1()

throws

InterruptedException

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask1

"

+

System.currentTimeMillis());

}

@Scheduled(cron

=

"0/1

*

*

*

*

?")

@Async//未指定線程池，則使用默認線程池

public

void

scheduledTask2()

{

System.out.println(Thread.currentThread().getName()

+

"---scheduledTask2

"

+

System.currentTimeMillis());

}

}控制臺執(zhí)行結(jié)果如下：從圖中依據(jù)線程名字，看到任務(wù)1均有自定義線程池defineAsyncTask-*執(zhí)行，同時驗證默認線程池SimpleAsyncTaskExecutor一直創(chuàng)建新線程執(zhí)行。2.定義全局線程池上面需在@Async()注解中指定使用自定義線程池才有效，如果我們即不想指定線程池，又不想使用默認線程池池—全局線程池。定義全局線程池可以通過實現(xiàn)AsyncConfigurer或者繼承AsyncConfigurerSupport。源碼如下：@Configuration

public

class

AsyncGlobalConfig

extends

AsyncConfigurerSupport

{

private

static

final

String

THREAD_PREFIX

=

"defineGlobalAsync-";

@Override

public

Executor

getAsyncExecutor()

{

ThreadPoolTaskExecutor

executor

=

new

ThreadPoolTaskExecutor();

executor.setThreadNamePrefix(THREAD_PREFIX);

executor.setCorePoolSize(3);

executor.setMaxPoolSize(10);

executor.setQueueCapacity(100);

executor.setKeepAliveSeconds(60);

executor.setRejectedExecutionHandler(new

ThreadPoolExecutor.CallerRunsPolicy());

executor.setWaitForTasksToCompleteOnShutdown(true);

executor.setAwaitTerminationSeconds(60);

executor.initialize();

return

executor;

}

}再次執(zhí)行上面的任務(wù)，結(jié)果如下：任務(wù)1，指定自定義線程池，則有該線程池執(zhí)行任務(wù)，其余未指定線程池，則使用自定義的全局線程池執(zhí)行任務(wù)。三、異常處理使用過線程池執(zhí)行任務(wù)的伙伴應(yīng)該會知道，線程提交任務(wù)分為execute()方式和submit()方式。對于異步submit提交任務(wù)時，使用Future.get()方法獲取返回結(jié)果時，主線程阻塞并可以處理線程池中的異常。對于execute()方式提交任務(wù)，當(dāng)異步任務(wù)返回類型為void，異常不會傳播到調(diào)用線程，故需要通過實現(xiàn)AsyncUncaughtExceptionHandler接口創(chuàng)建自定義異常處理。故在上面配置全局線程池的基礎(chǔ)上，處理異常。從源碼中可以得出AsyncConfigurerSupport提供了兩個方法，其中g(shù)etAsyncExecutor()是定義線程池的，getAsyncUncaughtExceptionHandler()是用于處理異常的。處理異常源碼實現(xiàn)如下：自定義異常處理實現(xiàn)類：static

class

CustomAsyncExceptionHandler

implements

AsyncUncaughtExceptionHandler

{

@Override

public

void

handleUncaughtExce