Spring Boot 實現異步任務執行 Async Task

Spring Boot 實現異步任務執行 Async Task

「異步」(Asynchronous)與「同步」(Synchronous)相對，異步不用阻塞當前線程來等待處理完成，而是容許後續操做，直至其它線程將處理完成，並回調通知此線程。也就是說，異步永遠是非阻塞的(non-blocking)。

同步操做的程序，會按照代碼的順序依次執行，每一行程序都必須等待上一個程序執行完成以後才能執行。哪些狀況建議使用同步交互呢？例如，銀行的轉帳系統，對數據庫的保存操做等等，都會使用同步交互操做。

異步操做的程序，在代碼執行時，不等待異步調用的語句返回結果就執行後面的程序。當任務間沒有前後順序依賴邏輯的時候，可使用異步。異步編程的主要困難在於，構建程序的執行邏輯時是非線性的，這須要將任務流分解成不少小的步驟，再經過異步回調函數的形式組合起來。

1.同步任務執行

下面經過一個簡單示例來直觀的理解什麼是同步任務執行。

編寫SyncTask類

編寫一個 SyncTask 類，建立三個處理函數：doTaskA()、doTaskB()、doTaskC() 來分別模擬三個任務執行的操做，操做消耗時間分別設置爲：1000ms、2000ms、3000ms。代碼以下

package com.easy.springboot.demo_async_task.task

import org.springframework.stereotype.Component

@Component
class SyncTask {

    fun doTaskA() {
        println("開始任務A")
        val start = System.currentTimeMillis()
        Thread.sleep(1000)
        val end = System.currentTimeMillis()
        println("結束任務A，耗時：" + (end - start) + "ms")
    }

    fun doTaskB() {
        println("開始任務B")
        val start = System.currentTimeMillis()
        Thread.sleep(2000)
        val end = System.currentTimeMillis()
        println("結束任務B，耗時：" + (end - start) + "ms")
    }

    fun doTaskC() {
        println("開始任務C")
        val start = System.currentTimeMillis()
        Thread.sleep(3000)
        val end = System.currentTimeMillis()
        println("結束任務C，耗時：" + (end - start) + "ms")
    }
}

單元測試

下面咱們來寫一個單元測試，在測試用例中順序執行 doTaskA()、doTaskB()、doTaskC() 三個函數。

package com.easy.springboot.demo_async_task

import com.easy.springboot.demo_async_task.task.AsyncTask
import com.easy.springboot.demo_async_task.task.SyncTask
import org.junit.Test
import org.junit.runner.RunWith
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.test.context.junit4.SpringRunner

@RunWith(SpringRunner::class)
@SpringBootTest
class DemoAsyncTaskApplicationTests {
    @Autowired lateinit var syncTask: SyncTask
    @Autowired lateinit var asyncTask: AsyncTask

    @Test
    fun testSyncTask() {
        println("開始測試SyncTask")
        val start = System.currentTimeMillis()
        syncTask.doTaskA()
        syncTask.doTaskB()
        syncTask.doTaskC()
        val end = System.currentTimeMillis()
        println("結束測試SyncTask，耗時：" + (end - start) + "ms")
    }

}

執行上面的單元測試，能夠在控制檯看到相似以下輸出：

開始測試SyncTask
開始任務A
結束任務A，耗時：1004ms
開始任務B
結束任務B，耗時：2005ms
開始任務C
結束任務C，耗時：3002ms
結束測試SyncTask，耗時：6012ms

任務A、任務B、任務C 依次按照其前後順序執行完畢，總共耗時：6012ms。

2.異步任務執行

上面的同步任務的執行，雖然順利的執行完了三個任務，但咱們能夠看到執行時間比較長，是這3個任務時間的累加。若這三個任務自己之間不存在依賴關係，能夠併發執行的話，同步順序執行在執行效率上就比較差了——這個時候，咱們能夠考慮經過異步調用的方式來實現「異步併發」地執行。

編寫AsyncTask類

在Spring Boot中，咱們只須要經過使用@Async註解就能簡單的將原來的同步函數變爲異步函數，編寫AsyncTask類，代碼改寫以下：

package com.easy.springboot.demo_async_task.task

import org.springframework.scheduling.annotation.Async
import org.springframework.scheduling.annotation.AsyncResult
import org.springframework.stereotype.Component
import java.util.*
import java.util.concurrent.Future

@Component
open class AsyncTask {

    @Async("asyncTaskExecutor")
    open fun doTaskA(): Future<String> {
        println("開始任務A")
        val start = System.currentTimeMillis()
        Thread.sleep(1000)
        val end = System.currentTimeMillis()
        println("結束任務A，耗時：" + (end - start) + "ms")
        return AsyncResult("TaskA DONE")
    }

    @Async("asyncTaskExecutor")
    open fun doTaskB(): Future<String> {
        println("開始任務B")
        val start = System.currentTimeMillis()
        Thread.sleep(2000)
        val end = System.currentTimeMillis()
        println("結束任務B，耗時：" + (end - start) + "ms")
        return AsyncResult("TaskB DONE")
    }

    @Async("asyncTaskExecutor")
    open fun doTaskC(): Future<String> {
        println("開始任務C")
        val start = System.currentTimeMillis()
        Thread.sleep(3000)
        val end = System.currentTimeMillis()
        println("結束任務C，耗時：" + (end - start) + "ms")
        return AsyncResult("TaskC DONE")
    }
}

上面的異步執行的任務，都返回一個 Future<String> 類型的結果對象 AsyncResult。這個對象中保存了任務的執行狀態。咱們能夠經過輪詢這個結果來等待任務執行完畢，這樣咱們能夠在上面3個任務都執行完畢後，再繼續作一些事情。

自定義線程池

其中，asyncTaskExecutor是咱們自定義的線程池。代碼以下

@Configuration
open class TaskExecutorPoolConfig {
    @Bean("asyncTaskExecutor")
    open fun taskExecutor(): Executor {
        val executor = ThreadPoolTaskExecutor()
        executor.corePoolSize = 10 //線程池維護線程的最少數量
        executor.maxPoolSize = 20 //線程池維護線程的最大數量
        executor.setQueueCapacity(100)
        executor.keepAliveSeconds = 30 //線程池維護線程所容許的空閒時間,TimeUnit.SECONDS
        executor.threadNamePrefix = "asyncTaskExecutor-"
        // 線程池對拒絕任務的處理策略: CallerRunsPolicy策略，當線程池沒有處理能力的時候，該策略會直接在 execute 方法的調用線程中運行被拒絕的任務；若是執行程序已關閉，則會丟棄該任務
        executor.setRejectedExecutionHandler(ThreadPoolExecutor.CallerRunsPolicy())
        return executor
    }
}

上面的代碼中，咱們經過使用ThreadPoolTaskExecutor建立了一個線程池，同時設置瞭如下這些參數，說明以下表：

核心線程數10：線程池建立時候初始化的線程數
最大線程數20：線程池最大的線程數，只有在緩衝隊列滿了以後纔會申請超過核心線程數的線程
緩衝隊列100：用來緩衝執行任務的隊列
容許線程的空閒時間30秒：當超過了核心線程出以外的線程在空閒時間到達以後會被銷燬
線程池名的前綴：設置好了以後能夠方便咱們定位處理任務所在的線程池
線程池對拒絕任務的處理策略：這裏採用了CallerRunsPolicy策略，當線程池沒有處理能力的時候，該策略會直接在 execute 方法的調用線程中運行被拒絕的任務；若是執行程序已關閉，則會丟棄該任務

啓用 @EnableAsync

爲了讓@Async註解可以生效，還須要在Spring Boot 的入口類上配置 @EnableAsync，代碼以下：

@SpringBootApplication
@EnableAsync
open class DemoAsyncTaskApplication

fun main(args: Array<String>) {
    runApplication<DemoAsyncTaskApplication>(*args)
}

單元測試

一樣地，咱們來編寫一個單元測試用例來測試一下異步執行這3個任務所花費的時間。代碼以下

package com.easy.springboot.demo_async_task

import com.easy.springboot.demo_async_task.task.AsyncTask
import com.easy.springboot.demo_async_task.task.SyncTask
import org.junit.Test
import org.junit.runner.RunWith
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.boot.test.context.SpringBootTest
import org.springframework.test.context.junit4.SpringRunner

@RunWith(SpringRunner::class)
@SpringBootTest
class DemoAsyncTaskApplicationTests {
    @Autowired lateinit var syncTask: SyncTask
    @Autowired lateinit var asyncTask: AsyncTask

    @Test
    fun testAsyncTask() {
        println("開始測試AsyncTask")
        val start = System.currentTimeMillis()

        val r1 = asyncTask.doTaskA()
        val r2 = asyncTask.doTaskB()
        val r3 = asyncTask.doTaskC()

        while (true) {
            // 三個任務都調用完成，退出循環等待
            if (r1.isDone && r2.isDone && r3.isDone) {
                break
            }
            Thread.sleep(100)
        }

        val end = System.currentTimeMillis()
        println("結束測試AsyncTask，耗時：" + (end - start) + "ms")
    }

}

咱們使用了一個死循環來等待三個任務都調用完成，當知足條件 r1.isDone && r2.isDone && r3.isDone 就退出循環等待。
執行上面的測試代碼，能夠在控制檯看到相似以下輸出

開始測試AsyncTask
開始任務A
開始任務B
開始任務C
結束任務A，耗時：1002ms
結束任務B，耗時：2004ms
結束任務C，耗時：3004ms
結束測試AsyncTask，耗時：3125ms

咱們能夠看到，經過異步調用，任務A、任務B、任務C 異步執行完畢總共耗時： 3125ms。 相比於同步執行，無疑大大的減小了程序的總運行時間。

提示：本節實例工程源代碼 https://github.com/EasySpring...

---

參考資料：
http://blog.didispace.com/spr...