Kotlin Coroutine(協程): 三、了解協程

@

目錄

• 前言
• 一、協程上下文
  • 1.調度器
  • 2.給協程起名
  • 3.局部變量
• 二、啟動模式 CoroutineStart
• 三、異常處理
  • 1.異常測試
  • 2.CoroutineExceptionHandler
• 四、監督:
  • 1.SupervisorJob
  • 2.supervisorScope
• 總結

前言

上一篇, 我們已經講述了協程的基本用法, 這篇將從協程上下文, 啟動模式, 異常處理角度來了解協程的用法

一、協程上下文

我們先看一下 啟動協程構建函數; launch, async等 它們參數都差不多

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {

第一個參數: CoroutineContext 就是協程上下文.
第二個參數: CoroutineStart 時協程的啟動模式, 我們后面再說
第三個參數: 就是協程的執行代碼塊.


CoroutineContext: 是一個接口, 它可以包含 調度器, 攔截協程執行, 局部變量等.
里面有一個操作符重載函數:

public operator fun plus(context: CoroutineContext): CoroutineContext = ...省略...

所以,才能看到 兩個上下文元素相加; 例如: SupervisorJob() + Dispatchers.Main
沒錯, 這就是 MainScope() 定義的上下文;

//kotlin.coroutines.CoroutineContext
public fun MainScope(): CoroutineScope = ContextScope(SupervisorJob() + Dispatchers.Main)

當然, 我們也可以看見 協程作用域 + 上下文

//kotlinx.coroutines.CoroutineScope
public operator fun CoroutineScope.plus(context: CoroutineContext): CoroutineScope =
    ContextScope(coroutineContext + context)

不管怎么加, 反正都是合並協程上下文中的內容.

1.調度器

上一篇已經介紹過了, 我們再次貼這幾種調度器的區別:

調度器	意義
不指定	它從啟動了它的 CoroutineScope 中承襲了上下文
Dispatchers.Main	用於Android. 在UI線程中執行
Dispatchers.IO	子線程, 適合執行磁盤或網絡 I/O操作
Dispatchers.Default	子線程,適合 執行 cpu 密集型的工作
Dispatchers.Unconfined	從當前線程直接執行, 直到第一個掛起點

2.給協程起名

還記得線程別名嗎? 沒錯 它們差不多; 它也是協程上下文元素

CoroutineName("name"):

launch(CoroutineName("v1coroutine")){...}

但要獲取附帶協程別名的線程名, 還得加JVM參數: -Dkotlinx.coroutines.debug

3.局部變量

有時，能夠將一些線程局部數據傳遞到協程與協程之間是很方便的。 它們不受任何特定線程的約束
使用 ThreadLocal 構建; 用 asContextElement(value = "launch") 轉換為協程上下文並賦值.

val threadLocal = ThreadLocal<String?>() // 聲明線程局部變量
runBlocking {
    threadLocal.set("main")
    letUsPrintln("start!! 變量值為:'${threadLocal.get()}';;")
    launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
        letUsPrintln("launch! 變量值為:'${threadLocal.get()}';;")
        delay(2000)
        launch{
            letUsPrintln("子協程! 變量值為:'${threadLocal.get()}';;")
        }
        letUsPrintln("launch! 變量值為:'${threadLocal.get()}';;")
    }
    launch {
        delay(1000)
        letUsPrintln("弟協程! 變量值為:'${threadLocal.get()}';;")
    }
    threadLocal.set(null)
    letUsPrintln("在末尾! 變量值為:'${threadLocal.get()}';;")
}

打印結果如下:

start!! 變量值為:'main';; Thread_name：main
launch! 變量值為:'launch';; Thread_name：DefaultDispatcher-worker-1
在末尾! 變量值為:'null';; Thread_name：main
弟協程! 變量值為:'null';; Thread_name：main
launch! 變量值為:'launch';; Thread_name：DefaultDispatcher-worker-1
子協程! 變量值為:'launch';; Thread_name：DefaultDispatcher-worker-1

注意:
當一個線程局部變量變化時，這個新值不會傳播給協程調用者

當然還有:
攔截器(ContinuationInterceptor): 多用作線程切換, 有興趣的小伙伴自行百度.
異常處理器(CoroutineExceptionHandler): 這個后面再說

二、啟動模式 CoroutineStart

1.DEFAULT
默認模式, 立即執行; 雖說立即執行, 實際上是立即調度執行. 代碼塊是否接着執行 還得看線程的空閑狀態啥的.

2.LAZY
延遲啟動, 我們可以先把協程定義好. 在需要的時候調用 start()

下面我們用 async 為例:

suspend fun doSomethingUsefulOne(): Int {
    println("doSomethingUsefulOne")
    delay(1000L) // 假設我們在這里做了些有用的事
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    println("doSomethingUsefulTwo")
    delay(500L) // 假設我們在這里也做了一些有用的事
    return 29
}

runBlocking {
    val one = async(start = CoroutineStart.LAZY) { doSomethingUsefulOne() }
    val two = async(start = CoroutineStart.LAZY) { doSomethingUsefulTwo() }
    delay(2000)	//掛起一下, 看看 LAZY 協程是否被啟動
    println("終於要啟動了")
    one.start() // 啟動第一個
    two.start() // 啟動第二個
    println("The answer is ${one.await() + two.await()}")
}

打印結果:

終於要啟動了
doSomethingUsefulOne
doSomethingUsefulTwo
The answer is 42

可以看出, 即使 delay(2000); LAZY模式的協程, 仍沒有啟動. 調用 start() 后才會啟動.

需要注意:
start() 或 await() 雖然都可以讓 LAZY協程啟動, 但上面的例子中, 只調用 await()的話, 兩個async會變為順序執行, 損失異步性質. 因此請使用 start() 來啟動 LAZY協程

3.ATOMIC
跟 DEFAULT 差不多, 區別在於 開始運行之前無法取消

如果不是 LAZY模式, 從協程定義 到代碼塊執行還是很簡短的. 這段時間內的取消與否 只能說也許在特殊業務中它才會被使用.


4.UNDISPATCHED
當前線程立即執行協程體，直到第一個掛起點.

怎么聽起來這么耳熟呢? 沒錯 它跟 調度器:Dispatchers.Unconfined 效果類似. 實現方式是否一致不得而知.

三、異常處理

異常處理較為復雜, 注意點也比較多, 真正理解需要很多測試代碼,或一定實戰經驗. 所以不能貫通理解也沒有關系,我們只需要對它有一定了解, 做到大體心中有數即可.

子協程:

我們先來了解一下子協程的定義:
當一個協程被其它協程在 CoroutineScope 中啟動的時候， 它將通過 CoroutineScope.coroutineContext 來承襲上下文，並且這個新協程的 Job 將會成為父協程作業的子作業。當一個父協程被取消的時候，所有它的子協程也會被遞歸的取消。
然而，當使用 GlobalScope 來啟動一個協程時，則新協程的作業沒有父作業。 因此它與這個啟動的作用域無關且獨立運作。一個父協程總是等待所有的子協程執行結束。父協程並不顯式的跟蹤所有子協程的啟動，並且不必使用 Job.join 在最后的時候等待它們

簡而言之:

• 協程中啟動的協程, 就是子協程. GlobalScope 除外; 新協程的Job, 也是子Job
• 父協程取消時(主動取消或異常取消), 遞歸取消所有子協程, 及子子協程
• 父協程會等待子協程全部執行完畢才會結束

當一個協程由於異常而運行失敗時:

1. 取消它自己的子級；
2. 取消它自己；
3. 將異常傳播並傳遞給它的父級。

異常會到達層級的根部，而且當前 CoroutineScope 所啟動的所有協程都會被取消。

1.異常測試

我們用幾個例子來檢測一下

runBlocking {
    launch {
        println("協程1-start")    //2
        delay(100)
        throw Exception("Failed coroutine") //4
    }
    launch {
        println("協程2-start")    //3
        delay(200)
        println("協程2-end")  //未打印
    }
    println("start")    //1
    delay(500)
    println("end")      //未打印
}

打印結果如下:

 start
 協程1-start
 協程2-start
 Exception in thread "main" java.lang.Exception: Failed coroutine ...

可以看出: 協程1異常. 協程2(兄弟協程)被取消. runBlocking(作用域)也被取消.


當 async 被用作根協程時，它的結果和異常會包裝在 返回值 Deferred.await() 中;

runBlocking {
    //async 依賴用戶來最終消費異常; 通過 await()
    val deferred = GlobalScope.async {
        letUsPrintln("協程1")
        throw Exception("Failed coroutine")
    }
    try {
        deferred.await()
    }catch (e: Exception){
        println("捕捉到了協程1異常")
    }
    letUsPrintln("end")
}

因此, try{..}catch {..} 需要包裹 await(); 而包裹 async{..} 是沒有意義的.

然而 try{..}catch{..} 並不一定合適;

runBlocking {
    try {
        launch {
            letUsPrintln("協程1")
            throw Exception("Failed coroutine")
        }
    }catch (e: Exception){
        println("捕捉到了協程1異常")	//未打印
    }
    delay(100)
    letUsPrintln("end")	//未打印
}

打印結果:

協程1 Thread_name：main
Exception in thread "main" java.lang.Exception: Failed coroutine ...

未能捕獲異常, runBlocking(父協程) 被終止; 我們嘗試用真實環境,包裹根協程:

try {
    lifecycleScope.launch {
        letUsPrintln("111協程1")
        throw Exception("Failed coroutine")
    }
}catch (e: Exception){
    println(e.message)
}

好吧, 程序直接 crash; 想想也對, 協程塊代碼始終是要分發給線程去做. try catch 又不是包在代碼塊里面.

2.CoroutineExceptionHandler

異常處理器, 它是 CoroutineContext 的一個可選元素，它讓您可以處理未捕獲的異常。

我們先定義一個 handler

val handler = CoroutineExceptionHandler {
    context, exception -> println("Caught $exception")
}

然后:

runBlocking {
    val scope = CoroutineScope(Job())	//自定義一個作用域
    val job = scope.launch(handler) {
        letUsPrintln("one")
        throw Exception("Failed coroutine")
    }
    job.join()
    letUsPrintln("end")
}

打印結果如下:

one Thread_name：DefaultDispatcher-worker-1
Caught java.lang.Exception: Failed coroutine
end Thread_name：main

這里新建作用域的目的, 是防止 launch 作為 runBlocking 的子協程; 我們去掉自定義作用域:

runBlocking {
    val job = launch(handler) {
        letUsPrintln("one")
        throw Exception("Failed coroutine")
    }
    job.join()
    letUsPrintln("end")	//未打印
}

打印結果如下:

one Thread_name：main
Exception in thread "main" java.lang.Exception: Failed coroutine ...

沒有捕獲異常, crash了. 這是為什么呢?

可以向上取消的子協程(非supervisor) 會委托父協程處理它們的異常. 所以異常是交給父協程處理. 而CoroutineExceptionHandler只能處理未被處理的異常, 因此:

• 把它加到 根協程 或作用域上. runBlocking,coroutineScope 中創建的協程不是根協程
• 單向取消的子協程(例如: supervisorScope 下的一級子協程), 這樣寫: launch(handler), 可以捕獲異常
• 其他情況, 子協程即便帶上Handler, 它也不生效

所以這樣可以捕獲異常:

lifecycleScope.launch(handler) {	//根協程 成功捕獲異常
    letUsPrintln("111協程1")
    throw Exception("Failed coroutine")
}

這樣無法捕獲異常:

lifecycleScope.launch {
    letUsPrintln("111協程1")
    launch(handler) {	//不能捕獲異常, 並引發 crash
        throw Exception("Failed coroutine")
    }
}

異常聚合:

當協程的多個子協程因異常而失敗時， 一般規則是“取第一個異常”，因此將處理第一個異常。 在第一個異常之后發生的所有其他異常都作為被抑制的異常綁定至第一個異常。

runBlocking {
    val handler = CoroutineExceptionHandler { _, exception ->
        println("CoroutineExceptionHandler got $exception with suppressed ${exception.suppressed.contentToString()}")
    }
    val job = GlobalScope.launch(handler) {
        launch {
            try {
                delay(Long.MAX_VALUE) // 當另一個同級的協程因 IOException  失敗時，它將被取消
            } finally {
                throw ArithmeticException() // 第二個異常
            }
        }
        launch {
            delay(100)
            throw IOException() // 首個異常
        }
        delay(Long.MAX_VALUE)
    }
    job.join()
}

打印結果只有一句, 如下所示:

CoroutineExceptionHandler got java.io.IOException with suppressed [java.lang.ArithmeticException]

結論:

CoroutineExceptionHandler : 以下稱之為 Handler

• async異常 依賴用戶調用 deferred.await(); 因此 Handler 在 async 這類協程構造器中無效;
• 當子協程的取消可以向上傳遞時(非supervisor類), Handler 只能加到 根協程 或作用域上, 子協程即便帶上Handler, 它也不生效
• CoroutineExceptionHandler 將等到所有子協程運行結束后再回調, 在收尾工作完成后.
• 它只是獲得異常信息. 拋出異常時, 協程將會遞歸終止, 並且無法通過 Handler 恢復.
• Handler 並不能恢復異常, 如果想捕獲異常, 並使協程繼續執行, 則應當使用 try{..}catch{..}

如下所示, try{..}catch{..} 放到協程體內部, 捕獲最初的異常本體:

launch {
    try {
    	// do something
    	throw ArithmeticException()	// 假定這里是可能拋異常的正常代碼
        delay(Long.MAX_VALUE) // 當另一個同級的協程因 IOException  失敗時，它將被取消
    } catch (e: ArithmeticException){
        // do something
    }
}

四、監督:

我們知道, 當子協程異常時, 會連帶父協程取消,直至取消整個作用域. 有時我們並不想要這樣, 例如 UI 作用域被取消, 導致其他正常的UI操作不能執行. 因此我們需要讓異常只向后傳遞.

1.SupervisorJob

使用 SupervisorJob 時，子協程的運行失敗不會影響到其他子協程。也不會傳播異常給它的父級，它會讓子協程自己處理異常。

runBlocking {
    val supervisor = SupervisorJob() //取消單向傳遞的 job
    with(CoroutineScope(coroutineContext + supervisor)) {
        launch {	//兄弟協程
            delay(100)
            println("第一個協程執行完畢")
        }
        launch {	//第二個協程拋出異常;
            throw AssertionError("The second child is cancelled")
        }
        delay(300)
        println("作用域被取消沒?")
    }
    println("全部執行完畢")
}

打印結果如下:

Exception in thread "main" java.lang.AssertionError: The first child is cancelled ...
第一個協程執行完畢
作用域被取消沒?
全部執行完畢

可以看出, 異常打印后. 兄弟協程 及 作用域都沒有被取消; 我們去掉 supervisor 再運行, 發現作用域協程被取消了. 可見是 SupervisorJob() 起了作用.

2.supervisorScope

對於作用域的並發，可以用 supervisorScope 來替代 coroutineScope 來實現相同的目的。它的直接子協程 將不會傳播異常給它的父級.

runBlocking {
    supervisorScope {
        launch {	//兄弟協程
            delay(100)
            println("第一個協程執行完畢")
        }
        launch {	//第二個協程拋出異常;
            throw AssertionError("The second child is cancelled")
        }
        delay(300)
        println("作用域被取消沒?")
    }
    println("全部執行完畢")
}

打印結果跟使用 with(CoroutineScope(coroutineContext + supervisor)) 時完全一致;


越級子協程

子子協程會不會將異常向上傳遞呢?

runBlocking {
	val scope = CoroutineScope(SupervisorJob())
    scope.launch {	//兄弟協程
        delay(100)
        println("第一個協程執行完畢")
    }
    scope.launch {	//協程二
        launch {    //第二個協程 的子協程 拋出異常;
            throw AssertionError("The second child is cancelled")
        }
        delay(200)
        println("第二個協程執行完畢?")	//未打印
    }
    delay(300)
    println("全部執行完畢")
}

打印結果如下:

Exception in thread "main" java.lang.AssertionError: The first child is cancelled ...
第一個協程執行完畢
全部執行完畢

可見, 第二個協程的完畢信息 未打印; 協程二 被取消; 這是因為監督只能作用一層, 它的直接子協程不會向上傳遞取消. 但子協程的內部還是普通的雙向傳遞模式;

小結:

• supervisorScope 會創建一個子作用域 (使用一個 SupervisorJob 作為父級); 以SupervisorJob 為父級的協程, 不會將取消操作向上級傳遞.
• SupervisorJob 只有作為 supervisorScope 或 CoroutineScope(SupervisorJob()) 的一部分時，才會按照上面的描述工作。

SupervisorJob() 的使用,一定是配合作用域(CoroutineScope) 的創建; 但當它作為參數傳入一個協程的 Builder 時 會怎么樣?:

runBlocking {
    val handler = CoroutineExceptionHandler { _, exception -> println("Caught $exception")}
    val jobBase = SupervisorJob()
    launch(jobBase) {	//與異常協程同一父 job;
        delay(50)
        println("協程1 執行完畢")
    }
    launch {	//新建 Job 承襲 父Job
        delay(60)
        println("協程2 執行完畢")
    }
    launch {	//新建 Job 承襲 父Job
        delay(70)
        println("協程3 執行完畢")
    }
    launch(jobBase+handler) {	//新建 Job 承襲 jobBase
        throw AssertionError("The first child is cancelled")
    }
    delay(100)
    println("全部執行完畢")
}

打印結果如下:

Caught java.lang.AssertionError: The first child is cancelled
協程1 執行完畢
協程2 執行完畢
協程3 執行完畢
全部執行完畢

這種方式, 實際上是替換了本該從父協程中承襲的Job;
可見 同父Job的 協程1 並沒有被取消; 我們換成 Job 試試; 只需要更換一句代碼:

val jobBase = Job()

結果如下:

Caught java.lang.AssertionError: The first child is cancelled
協程2 執行完畢
協程3 執行完畢
全部執行完畢

可見 同父Job的 協程1 被取消; 協程2和協程3正常執行;

注意: 這種直接將Job傳入協程Builder 的方式, 會破壞原本協程繼承 Job的模式;

總結

CoroutineContext 協程上下文;

• 調度器: 四種調度器, 可以指定協程的執行方式, 或執行線程
• 還有協程別名, 局部變量, 攔截器, 異常處理器等

CoroutineStart 啟動模式

• 四種啟動模式, 延遲啟動等

異常處理:

• CoroutineExceptionHandler: 處理未被處理的異常
• 監督: 一般配合創建作用域 CoroutineScope(SupervisorJob()); 或使用 supervisorScope;

注意點:

• 當一個協程由於異常而運行失敗時, 會取消所有子協程, 取消自己, 再傳播給父級, 直到取消整個作用域,
• 異常處理器只能處理 未被處理的異常, 在雙向取消的子協程中不起作用. 在 async 類協程中不起作用
• 監督: 會在作用域內 使用一個SupervisorJob作為父級. 只能生效一層. 因為子協程會新建自己的Job, 子子協程繼承的是 Job, 而不是 SupervisorJob
• 當 async 不是根協程時, 異常仍然會通過 Job 向上傳遞, 導致作用域取消, crash等; runBlocking, coroutineScope 的代碼塊中創建的協程, 並不是根協程