Kotlin協(xié)程Job生命周期結(jié)構(gòu)化并發(fā)詳解

更新時間：2022年12月08日 08:40:18 作者：無糖可樂愛好者

這篇文章主要為大家介紹了Kotlin協(xié)程Job生命周期結(jié)構(gòu)化并發(fā)詳解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步，早日升職加薪

引言

前面在學(xué)習(xí)協(xié)程啟動方式的時候在launch的源碼中有一個返回值是Job，async的返回Deferred也是實現(xiàn)了Job，那么而也就是說launch和async在創(chuàng)建一個協(xié)程的時候也會創(chuàng)建一個對應(yīng)的Job對象。還提到過Job是協(xié)程的句柄，那么Job到底是什么？它有什么用？

1.Job的生命周期

先看一下Job的源碼，這里只保留了跟標題相關(guān)的內(nèi)容

public interface Job : CoroutineContext.Element {
    // ------------ 狀態(tài)查詢API ------------
    /**
    * 當該Job處于活動狀態(tài)時，返回true——它已經(jīng)開始，沒有完成，也沒有取消。
    * 如果沒有取消或失敗，等待其子任務(wù)完成的Job仍被認為是活動的。
    */
    public val isActive: Boolean
    /**
    * 當Job因任何原因完成時返回true。作業(yè)被取消或失敗并已完成其執(zhí)行也被視為完成。
    * Job只有在所有子任務(wù)完成后才算完成。
    */
    public val isCompleted: Boolean
    /**
    *如果該作業(yè)因任何原因被取消，無論是通過顯式調(diào)用cancel，還是因為它失敗或其子或父作業(yè)被取消，
    * 則返回true。在一般情況下，它并不意味著任務(wù)已經(jīng)完成，因為它可能仍然在完成它正在做的事情，
    * 并等待它的子任務(wù)完成。
    */
    public val isCancelled: Boolean
    // ------------ 操控狀態(tài)API ------------
    /**
    * 如果Job所在的協(xié)程還沒有被啟動那么調(diào)用這個方法就會啟動協(xié)程
    * 如果這個協(xié)程被啟動了返回true，如果已經(jīng)啟動或者執(zhí)行完畢了返回false
    */
    public fun start(): Boolean
    /**
    * 取消此Job，可用于指定錯誤消息或提供有關(guān)取消原因的其他詳細信息
    */
    public fun cancel(cause: CancellationException? = null)
    /**
    * 取消此Job
    */
    public fun cancel(): Unit = cancel(null)
    public fun cancel(cause: Throwable? = null): Boolean
    // ------------ 等待狀態(tài)API ------------
    /**
    * 掛起協(xié)程，知道任務(wù)完成再恢復(fù)
    */
    public suspend fun join()
    // ------------ 完成狀態(tài)回調(diào)API ------------
    /**
    * 注冊Job完成時同步調(diào)用的處理程序.
    * 當Job已經(jīng)完成時，將處理程序?qū)⒘⒓凑{(diào)用Job的異?；蛉∠蚧騨ull
    * 否則，該處理程序?qū)⒃诖薐ob完成時調(diào)用一次。
    */
    public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
    /**
    * 注冊在取消或完成此Job時同步調(diào)用的處理程序。
    * 當Job已經(jīng)被取消并完成執(zhí)行時，處理程序?qū)⒘⒓凑{(diào)用Job的取消原因或null，
    * 除非將invokeImmediately設(shè)置為false。否則，
    * 當Job取消或完成時將調(diào)用一次handler。
    */
    public fun invokeOnCompletion(
        onCancelling: Boolean = false,
        invokeImmediately: Boolean = true,
        handler: CompletionHandler): DisposableHandle
}

從源碼中可以發(fā)現(xiàn)這幾個函數(shù)和變量跟Actviity或者Fragment非常像，所以我們可以總結(jié)出兩個結(jié)論：

Job可以監(jiān)測協(xié)程的生命周期
Job可以操控協(xié)程

在例子中使用這幾個函數(shù)和變量再來校驗一下上面的結(jié)論：

fun main() = runBlocking {
    val job = launch {
        delay(1000L)
    }
    job.log()
    job.cancel()
    job.log()
}
fun Job.log() {
    println(
        """
        isActive：$isActive
        isCompleted：$isCompleted
        isCancelled：$isCancelled
        Thread：${Thread.currentThread().name}  
        ================================
    """.trimIndent()
    )
}
//輸出結(jié)果
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：false
//isCompleted：false
//isCancelled：true
//Thread：main @coroutine#1  
//================================

Job.log用了擴展函數(shù)，方便調(diào)用Job中的狀態(tài)監(jiān)測返回值。

上面的代碼通過launch創(chuàng)建了一個協(xié)程，接收了Job的返回值，這里用這個job對象做了三件事：

第一個job.log() ：launch的創(chuàng)建標志著協(xié)程已經(jīng)被啟動所以在第一個job.log()的日志中isActive返回值是true；
job.cancel() ：這里調(diào)用了job的取消函數(shù)將協(xié)程任務(wù)取消；
第二個job.log() ：上面的代碼將協(xié)程任務(wù)取消了，然后再次獲取協(xié)程狀態(tài)發(fā)現(xiàn)isActivte返回false，isCancelled返回true。

上面的代碼也印證了前面提出的結(jié)論，還有一個函數(shù)start沒使用，再來調(diào)用它之后輸出的日志：

fun main() = runBlocking {
    //變化1
    val job = launch(start = CoroutineStart.LAZY) {
        delay(1000L)
    }
    job.log()
    //變化2
    job.start()
    job.log()
    job.cancel()
    job.log()
}
fun Job.log() {
    println(
        """
        isActive：$isActive
        isCompleted：$isCompleted
        isCancelled：$isCancelled
        Thread：${Thread.currentThread().name}  
        ================================
    """.trimIndent()
    )
}
//輸出結(jié)果：
//isActive：false
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：false
//isCompleted：false
//isCancelled：true
//Thread：main @coroutine#1  
//================================

上面的代碼增加了兩處修改：

變化1：協(xié)程在創(chuàng)建出來的時候就已經(jīng)被啟動，因此為了查看調(diào)用Job.start()前的日志需要加上懶啟動
變化2：調(diào)用start函數(shù)啟動協(xié)程

從輸出結(jié)果來看沒有調(diào)用start函數(shù)前isActive返回true，調(diào)用后就返回了true，當使用懶啟動后在調(diào)用cancel函數(shù)與前面使用cancel函數(shù)輸出的日志是一樣的，可以得知懶啟動后對協(xié)程的生命周期并沒有設(shè)么影響(這可能是句廢話)。

現(xiàn)在還有最后一個變量沒有看isCompleted，在上面的代碼中添加一個延時函數(shù)，等協(xié)程任務(wù)結(jié)束再打印日志

fun main() = runBlocking {
    val job = launch(start = CoroutineStart.LAZY) {
        delay(1000L)
    }
    job.log()
    job.start()
    job.log()
    job.cancel()
    delay(2000L)		//變化在這里
    job.log()
}
fun Job.log() {
    println(
        """
        isActive：$isActive
        isCompleted：$isCompleted
        isCancelled：$isCancelled
        Thread：${Thread.currentThread().name}  
        ================================
    """.trimIndent()
    )
}
//輸出結(jié)果：
//isActive：false
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：false
//isCompleted：true
//isCancelled：true
//Thread：main @coroutine#1  
//================================

從輸出結(jié)果中看到當調(diào)用isCancel后isCompleted也返回了true，也就是說任務(wù)結(jié)束了。

上面的代碼為了監(jiān)測isCompleted的狀態(tài)加了一個延時函數(shù)delay，但是這種方式并不建議使用，因為這個時間他不是固定的，例如從后臺請求數(shù)據(jù)或者下載文件，這種情況下的時間是完全無法預(yù)知的。

現(xiàn)在假設(shè)已經(jīng)知道協(xié)程執(zhí)行完畢需要delay(1000L)的時間，如果將協(xié)程內(nèi)的delay時長設(shè)置的大于外部的delay時長，會帶來什么問題？

fun main() = runBlocking {
    val job = launch(start = CoroutineStart.LAZY) {
        delay(4000L)
    }
    job.log()
    job.start()
    job.log()
    delay(1000L)
    job.log()
    println("Process end!")
}
fun Job.log() {
    println(
        """
        isActive：$isActive
        isCompleted：$isCompleted
        isCancelled：$isCancelled
        Thread：${Thread.currentThread().name}  
        ================================
    """.trimIndent()
    )
}
//輸出結(jié)果：
//isActive：false
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//Process end!

由輸出結(jié)果可知isCompleted狀態(tài)是false，協(xié)程任務(wù)是否執(zhí)行完畢不得而知。另外當println("Process end!")執(zhí)行完畢后程序并沒有立即輸出Process finished with exit code 0，這是因為runBlocking 會一直阻塞，等到 job 任務(wù)執(zhí)行完畢以后才真正退出。

那要如何解決這個問題？

//Job#join 
/**
* 掛起協(xié)程，知道任務(wù)完成再恢復(fù)
*/
public suspend fun join()

join是Job中的一個掛起函數(shù)，調(diào)用后會掛起當前程序的執(zhí)行流程，等待job當中的協(xié)程任務(wù)執(zhí)行完畢然后再恢復(fù)當前程序的執(zhí)行流程。

join將任務(wù)掛起后再恢復(fù)，那要如何知道任務(wù)是否執(zhí)行完畢了？invokeOnCompletion可以監(jiān)聽任務(wù)執(zhí)行的狀態(tài)

//Job#invokeOnCompletion
/**
* 注冊Job完成時同步調(diào)用的處理程序.
* 當Job已經(jīng)完成時，將處理程序?qū)⒘⒓凑{(diào)用Job的異常或取消原因或null
* 否則，該處理程序?qū)⒃诖薐ob完成時調(diào)用一次。
*/
public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
//Job#invokeOnCompletion
/**
* 注冊在取消或完成此Job時同步調(diào)用的處理程序。
* 當Job已經(jīng)被取消并完成執(zhí)行時，處理程序?qū)⒘⒓凑{(diào)用Job的取消原因或null，
* 除非將invokeImmediately設(shè)置為false。否則，
* 當Job取消或完成時將調(diào)用一次handler。
*/
public fun invokeOnCompletion(
        onCancelling: Boolean = false,
        invokeImmediately: Boolean = true,
        handler: CompletionHandler): DisposableHandle

join和invokeOnCompletion的使用如下：

fun main() = runBlocking {
    val job = launch(start = CoroutineStart.LAZY) {
        delay(4000L)
    }
    job.log()
    job.start()
    job.log()
    //新增
    job.join()
    //新增
    job.invokeOnCompletion {
        println("==========Task status==========")
        job.log()
    }
    println("Process end!")
}
fun Job.log() {
    println(
        """
        isActive：$isActive
        isCompleted：$isCompleted
        isCancelled：$isCancelled
        Thread：${Thread.currentThread().name}  
        ================================
    """.trimIndent()
    )
}
//輸出結(jié)果：
//isActive：false
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//isActive：true
//isCompleted：false
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//==========Task status==========
//isActive：false
//isCompleted：true
//isCancelled：false
//Thread：main @coroutine#1  
//================================
//Process end!

可以看到加入join和invokeOnCompletion之后isCompleted的狀態(tài)就正確了，同時Process end!輸出后Process finished with exit code 0也會很快的輸出，這說明任務(wù)確實執(zhí)行完畢了。

在講協(xié)程的啟動方式的時候提出一個觀點：launch的返回值Job代表的是協(xié)程的句柄。那么Job是協(xié)程的句柄該怎么理解？

句柄： 是指一個中間媒介，可以操控一個東西。就類似于遙控器操作空調(diào)場景中遙控器就是句柄，開關(guān)控制燈具場景中開關(guān)就是句柄。

所以Job和協(xié)程的關(guān)系就類似于遙控器和空調(diào)，開關(guān)和燈具。Job可以監(jiān)測協(xié)程的運行狀態(tài)也可以控制協(xié)程的運行狀態(tài)。那么Job就和遙控器、開關(guān)一樣看做是一個句柄。

2.Deffered

launch直接創(chuàng)建了Job，async通過Deffered間接創(chuàng)建了Job對象，但是它并沒有在 Job 的基礎(chǔ)上擴展出很多其他功能，而接收一個返回值是依靠 await() 方法，那await方法是如何實現(xiàn)的？

fun main() = runBlocking {
    val deferred = async {
        logX("Coroutine start!")
        delay(1000L)
        logX("Coroutine end!")
        "Coroutine result!"
    }
    val result = deferred.await()
    println("Result = $result")
    logX("Process end!")
}
fun logX(any: Any?) {
    println(
        """
================================
$any 
Thread:${Thread.currentThread().name}
================================
""".trimIndent()
    )
}
//輸出結(jié)果：
//Coroutine start! 
//Thread:main @coroutine#2
//================================
//================================
//Coroutine end! 
//Thread:main @coroutine#2
//================================
//Result = Coroutine result!
//================================
//Process end! 
//Thread:main @coroutine#1

從輸出結(jié)果來看，await方法可以獲取協(xié)程執(zhí)行結(jié)果外，好像還會阻塞協(xié)程的執(zhí)行流程，直到協(xié)程任務(wù)執(zhí)行完畢。看一下await的源碼

//Deferred#await
public interface Deferred<out T> : Job {
    ...
    public suspend fun await(): T
    ...
}

從源碼來看await也是一個掛起函數(shù)，它跟join是一樣的，看似阻塞的過程其實是協(xié)程的掛起和恢復(fù)能力。

所以，總的來說，Deferred 只是比 Job 多了一個 await() 掛起函數(shù)而已，通過這個掛起函數(shù)，就可以等待協(xié)程執(zhí)行完畢的同時，還可以直接拿到協(xié)程的執(zhí)行結(jié)果。

3.Job與結(jié)構(gòu)化并發(fā)

在其他地方看過這么一句話：協(xié)程的優(yōu)勢在于結(jié)構(gòu)化并發(fā)， 這句話該如何理解？

這句話可以理解為帶有結(jié)構(gòu)和層級的并發(fā)，用代碼表現(xiàn)就像這樣：

fun main() = runBlocking {
    val parentJob: Job
    var childJob1: Job? = null
    var childJob2: Job? = null
    var childJob3: Job? = null
    parentJob = launch {
        childJob1 = launch {
            delay(1000L)
        }
        childJob2 = launch {
            delay(3000L)
        }
        childJob3 = launch {
            delay(5000L)
        }
    }
    delay(500L)
    parentJob.children.forEachIndexed { index, job ->
        when (index) {
            0 -> println("childJob1 === childJob1 is ${childJob1 === job}")
            1 -> println("childJob2 === childJob2 is ${childJob2 === job}")
            2 -> println("childJob3 === childJob3 is ${childJob3 === job}")
        }
    }
    parentJob.join()
    logX("Process end!")
}
//輸出結(jié)果：
//childJob1 === childJob1 is true
//childJob2 === childJob2 is true
//childJob3 === childJob3 is true
//================================
//Process end! 
//Thread:main @coroutine#1

上面的代碼是父子層級，父Job使用launch啟動了協(xié)程同時它的內(nèi)部還有三個Job，三個子Job是并發(fā)執(zhí)行的，同時也是用過launch啟動的協(xié)程，調(diào)用了parentJob.join()那么掛起的時間就是childJob3的時長—5秒，因為它要等待所有任務(wù)都執(zhí)行完畢才會恢復(fù)執(zhí)行，然后通過children.forEachIndexed進行遍歷并分別對比他們與三個子Job的引用是否相等“===”代表了引用相等，即是否是同一個對象）。圖示如下

前面講過，Job可以調(diào)用cancel方法取消執(zhí)行，那么當調(diào)用parentJob.cancel會有什么樣的情況？

fun main() = runBlocking {
    val parentJob: Job
    var childJob1: Job? = null
    var childJob2: Job? = null
    var childJob3: Job? = null
    parentJob = launch {
        childJob1 = launch {
            println("childJob1 start")
            delay(1000L)
            println("childJob1 end")
        }
        childJob2 = launch {
            println("childJob2 start")
            delay(3000L)
            println("childJob2 start")
        }
        childJob3 = launch {
            println("childJob3 start")
            delay(5000L)
            println("childJob3 start")
        }
    }
    delay(500L)
    parentJob.cancel()
    logX("Process end!")
}
//輸出結(jié)果：
//childJob1 start
//childJob2 start
//childJob3 start
//================================
//Process end! 
//Thread:main @coroutine#1

parentJob.cancel調(diào)用后，每個子Job只是輸出了start，這就可以得出一個結(jié)論：父Job取消后子Job也會依次跟著取消。如果調(diào)用任何一個子Job的cancel則不會對父Job和其他子Job產(chǎn)生影響。

到這里對于開頭的那句協(xié)程的優(yōu)勢在于結(jié)構(gòu)化并發(fā)就有更更好的理解了，這是Kotlin協(xié)程的第二大優(yōu)勢。

4.launch和async的使用場景

launch： 主要用來發(fā)起一些不需要任何結(jié)果的耗時任務(wù)，這個任務(wù)在執(zhí)行中可以改變它的執(zhí)行狀態(tài)。
async： 主要用來發(fā)起一些需要結(jié)果的耗時任務(wù)，以及與掛起函數(shù)結(jié)合，優(yōu)化并發(fā)。