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

更新時(shí)間：2022年12月08日 08:40:18 作者：無(wú)糖可樂(lè)愛(ài)好者

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

引言

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

1.Job的生命周期

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

public interface Job : CoroutineContext.Element {
    // ------------ 狀態(tài)查詢API ------------
    /**
    * 當(dāng)該Job處于活動(dòng)狀態(tài)時(shí)，返回true——它已經(jīng)開(kāi)始，沒(méi)有完成，也沒(méi)有取消。
    * 如果沒(méi)有取消或失敗，等待其子任務(wù)完成的Job仍被認(rèn)為是活動(dòng)的。
    */
    public val isActive: Boolean
    /**
    * 當(dāng)Job因任何原因完成時(shí)返回true。作業(yè)被取消或失敗并已完成其執(zhí)行也被視為完成。
    * Job只有在所有子任務(wù)完成后才算完成。
    */
    public val isCompleted: Boolean
    /**
    *如果該作業(yè)因任何原因被取消，無(wú)論是通過(guò)顯式調(diào)用cancel，還是因?yàn)樗』蚱渥踊蚋缸鳂I(yè)被取消，
    * 則返回true。在一般情況下，它并不意味著任務(wù)已經(jīng)完成，因?yàn)樗赡苋匀辉谕瓿伤谧龅氖虑椋?
    * 并等待它的子任務(wù)完成。
    */
    public val isCancelled: Boolean
    // ------------ 操控狀態(tài)API ------------
    /**
    * 如果Job所在的協(xié)程還沒(méi)有被啟動(dòng)那么調(diào)用這個(gè)方法就會(huì)啟動(dòng)協(xié)程
    * 如果這個(gè)協(xié)程被啟動(dòng)了返回true，如果已經(jīng)啟動(dòng)或者執(zhí)行完畢了返回false
    */
    public fun start(): Boolean
    /**
    * 取消此Job，可用于指定錯(cuò)誤消息或提供有關(guān)取消原因的其他詳細(xì)信息
    */
    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 ------------
    /**
    * 注冊(cè)Job完成時(shí)同步調(diào)用的處理程序.
    * 當(dāng)Job已經(jīng)完成時(shí)，將處理程序?qū)⒘⒓凑{(diào)用Job的異常或取消原因或null
    * 否則，該處理程序?qū)⒃诖薐ob完成時(shí)調(diào)用一次。
    */
    public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
    /**
    * 注冊(cè)在取消或完成此Job時(shí)同步調(diào)用的處理程序。
    * 當(dāng)Job已經(jīng)被取消并完成執(zhí)行時(shí)，處理程序?qū)⒘⒓凑{(diào)用Job的取消原因或null，
    * 除非將invokeImmediately設(shè)置為false。否則，
    * 當(dāng)Job取消或完成時(shí)將調(diào)用一次handler。
    */
    public fun invokeOnCompletion(
        onCancelling: Boolean = false,
        invokeImmediately: Boolean = true,
        handler: CompletionHandler): DisposableHandle
}

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

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

在例子中使用這幾個(gè)函數(shù)和變量再來(lái)校驗(yàn)一下上面的結(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用了擴(kuò)展函數(shù)，方便調(diào)用Job中的狀態(tài)監(jiān)測(cè)返回值。

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

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

上面的代碼也印證了前面提出的結(jié)論，還有一個(gè)函數(shù)start沒(méi)使用，再來(lái)調(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)建出來(lái)的時(shí)候就已經(jīng)被啟動(dòng)，因此為了查看調(diào)用Job.start()前的日志需要加上懶啟動(dòng)
變化2：調(diào)用start函數(shù)啟動(dòng)協(xié)程

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

現(xiàn)在還有最后一個(gè)變量沒(méi)有看isCompleted，在上面的代碼中添加一個(gè)延時(shí)函數(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é)果中看到當(dāng)調(diào)用isCancel后isCompleted也返回了true，也就是說(shuō)任務(wù)結(jié)束了。

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

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

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í)行完畢不得而知。另外當(dāng)println("Process end!")執(zhí)行完畢后程序并沒(méi)有立即輸出Process finished with exit code 0，這是因?yàn)閞unBlocking 會(huì)一直阻塞，等到 job 任務(wù)執(zhí)行完畢以后才真正退出。

那要如何解決這個(gè)問(wèn)題？

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

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

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

//Job#invokeOnCompletion
/**
* 注冊(cè)Job完成時(shí)同步調(diào)用的處理程序.
* 當(dāng)Job已經(jīng)完成時(shí)，將處理程序?qū)⒘⒓凑{(diào)用Job的異?；蛉∠蚧騨ull
* 否則，該處理程序?qū)⒃诖薐ob完成時(shí)調(diào)用一次。
*/
public fun invokeOnCompletion(handler: CompletionHandler): DisposableHandle
//Job#invokeOnCompletion
/**
* 注冊(cè)在取消或完成此Job時(shí)同步調(diào)用的處理程序。
* 當(dāng)Job已經(jīng)被取消并完成執(zhí)行時(shí)，處理程序?qū)⒘⒓凑{(diào)用Job的取消原因或null，
* 除非將invokeImmediately設(shè)置為false。否則，
* 當(dāng)Job取消或完成時(shí)將調(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)就正確了，同時(shí)Process end!輸出后Process finished with exit code 0也會(huì)很快的輸出，這說(shuō)明任務(wù)確實(shí)執(zhí)行完畢了。

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

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

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

2.Deffered

launch直接創(chuàng)建了Job，async通過(guò)Deffered間接創(chuàng)建了Job對(duì)象，但是它并沒(méi)有在 Job 的基礎(chǔ)上擴(kuò)展出很多其他功能，而接收一個(gè)返回值是依靠 await() 方法，那await方法是如何實(shí)現(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é)果來(lái)看，await方法可以獲取協(xié)程執(zhí)行結(jié)果外，好像還會(huì)阻塞協(xié)程的執(zhí)行流程，直到協(xié)程任務(wù)執(zhí)行完畢?？匆幌?code>await的源碼

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

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

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

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

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

這句話可以理解為帶有結(jié)構(gòu)和層級(jí)的并發(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

上面的代碼是父子層級(jí)，父Job使用launch啟動(dòng)了協(xié)程同時(shí)它的內(nèi)部還有三個(gè)Job，三個(gè)子Job是并發(fā)執(zhí)行的，同時(shí)也是用過(guò)launch啟動(dòng)的協(xié)程，調(diào)用了parentJob.join()那么掛起的時(shí)間就是childJob3的時(shí)長(zhǎng)—5秒，因?yàn)樗却腥蝿?wù)都執(zhí)行完畢才會(huì)恢復(fù)執(zhí)行，然后通過(guò)children.forEachIndexed進(jìn)行遍歷并分別對(duì)比他們與三個(gè)子Job的引用是否相等“===”代表了引用相等，即是否是同一個(gè)對(duì)象）。圖示如下

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

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)用后，每個(gè)子Job只是輸出了start，這就可以得出一個(gè)結(jié)論：父Job取消后子Job也會(huì)依次跟著取消。如果調(diào)用任何一個(gè)子Job的cancel則不會(huì)對(duì)父Job和其他子Job產(chǎn)生影響。

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

4.launch和async的使用場(chǎng)景

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