再談協(xié)程之 Suspend 到底掛起了啥

作者：徐宜生 2021-12-01 09:47:20

Kotlin編譯器會(huì)給每一個(gè)suspend函數(shù)生成一個(gè)狀態(tài)機(jī)來管理協(xié)程的執(zhí)行。

Coroutines簡(jiǎn)化了Android上的異步操作。正如文檔中所解釋的，我們可以用它們來管理異步任務(wù)，否則可能會(huì)阻塞主線程，導(dǎo)致你的應(yīng)用程序Crash。

Coroutines也有助于用命令式的代碼取代基于回調(diào)的API。

作為例子，我們先看看這個(gè)使用回調(diào)的異步代碼。

// Simplified code that only considers the happy path 
fun loginUser(userId: String, password: String, userResult: Callback<User>) { 
  // Async callbacks 
  userRemoteDataSource.logUserIn { user -> 
    // Successful network request 
    userLocalDataSource.logUserIn(user) { userDb -> 
      // Result saved in DB 
      userResult.success(userDb) 
    } 
  } 
}

這些回調(diào)可以使用coroutines轉(zhuǎn)換為順序的函數(shù)調(diào)用。

suspend fun loginUser(userId: String, password: String): User { 
  val user = userRemoteDataSource.logUserIn(userId, password) 
  val userDb = userLocalDataSource.logUserIn(user) 
  return userDb 
}

在coroutines代碼中，我們給函數(shù)添加了suspend修飾符。這將告訴編譯器，這個(gè)函數(shù)需要在一個(gè)coroutine內(nèi)執(zhí)行。作為一個(gè)開發(fā)者，你可以把suspend函數(shù)看作是一個(gè)普通的函數(shù)，但它的執(zhí)行可能被掛起，并在某個(gè)時(shí)候恢復(fù)。

簡(jiǎn)而言之，suspend就是一種編譯器生成的回調(diào)。

與回調(diào)不同的是，coroutines提供了一種在線程之間切換和處理異常的簡(jiǎn)單方法。

但是，當(dāng)我們把函數(shù)標(biāo)記為suspend時(shí)，編譯器實(shí)際上在幕后做了什么?

Suspend到底做了什么

回到loginUser的suspend函數(shù)，注意它調(diào)用的其他函數(shù)也是suspend函數(shù)。

suspend fun loginUser(userId: String, password: String): User { 
  val user = userRemoteDataSource.logUserIn(userId, password) 
  val userDb = userLocalDataSource.logUserIn(user) 
  return userDb 
} 
// UserRemoteDataSource.kt 
suspend fun logUserIn(userId: String, password: String): User 
// UserLocalDataSource.kt 
suspend fun logUserIn(userId: String): UserDb

簡(jiǎn)而言之，Kotlin編譯器將使用有限狀態(tài)機(jī)(我們將在后面介紹)把suspend函數(shù)轉(zhuǎn)換為優(yōu)化版本的回調(diào)實(shí)現(xiàn)。你說對(duì)了，編譯器會(huì)幫你寫這些回調(diào)，它們的本質(zhì)，依然是回調(diào)!

Continuation的真面目

suspend函數(shù)之間的通信方式是使用Continuation對(duì)象。一個(gè)Continuation只是一個(gè)帶有一些額外信息的通用回調(diào)接口。正如我們稍后將看到的，它將代表一個(gè)suspend函數(shù)的生成狀態(tài)機(jī)。

讓我們看一下它的定義。

interface Continuation<in T> { 
  public val context: CoroutineContext 
  public fun resumeWith(value: Result<T>) 
}

context是在continuation中使用的CoroutineContext。

resumeWith用一個(gè)Result來恢復(fù)Coroutine的執(zhí)行，這個(gè)Result可以包含一個(gè)導(dǎo)致suspend的計(jì)算結(jié)果的值或者是一個(gè)異常。

注意：從Kotlin 1.3開始，你還可以使用擴(kuò)展函數(shù)resume(value: T)和resumeWithException(exception: Throwable)，它們是resumeWith調(diào)用的特殊版本。

編譯器將使用函數(shù)簽名中的額外參數(shù)completion(Continuation類型)替換suspend修飾符，該參數(shù)將用于將suspend函數(shù)的結(jié)果傳達(dá)給調(diào)用它的coroutine。

fun loginUser(userId: String, password: String, completion: Continuation<Any?>) { 
  val user = userRemoteDataSource.logUserIn(userId, password) 
  val userDb = userLocalDataSource.logUserIn(user) 
  completion.resume(userDb) 
}

為了簡(jiǎn)單起見，我們的例子將返回Unit而不是User。User對(duì)象將在添加的Continuation參數(shù)中被 "返回"。

suspend函數(shù)的字節(jié)碼實(shí)際上返回 Any? 因?yàn)樗?(T | COROUTINE_SUSPENDED)的聯(lián)合類型。這允許函數(shù)在可以時(shí)同步返回。

注意：如果你用suspend修飾符標(biāo)記一個(gè)不調(diào)用其他suspend函數(shù)的函數(shù)，編譯器也會(huì)添加額外的Continuation參數(shù)，但不會(huì)對(duì)它做任何事情，函數(shù)體的字節(jié)碼看起來就像一個(gè)普通函數(shù)。

你也可以在其他地方看到Continuation接口。

當(dāng)使用suspendCoroutine或suspendCancellableCoroutine將基于回調(diào)的API轉(zhuǎn)換為coroutine時(shí)(你應(yīng)該總是傾向于使用這種方法)，你直接與Continuation對(duì)象交互，以恢復(fù)在運(yùn)行時(shí)被suspend的作為參數(shù)傳遞的代碼塊。

你可以使用suspend函數(shù)上的startCoroutine擴(kuò)展函數(shù)來啟動(dòng)一個(gè)coroutine。它接收一個(gè)Continuation對(duì)象作為參數(shù)，當(dāng)新的coroutine完成時(shí)，無論是結(jié)果還是異常，都會(huì)被調(diào)用。

切換不同的Dispatchers

你可以在不同的Dispatchers之間進(jìn)行交換，在不同的線程上執(zhí)行計(jì)算。那么Kotlin如何知道在哪里恢復(fù)一個(gè)暫停的計(jì)算?

Continuation有一個(gè)子類型，叫做DispatchedContinuation，它的resume函數(shù)可以對(duì)CoroutineContext中可用的Dispatcher進(jìn)行調(diào)度調(diào)用。除了Dispatchers.Unconfined的isDispatchNeeded函數(shù)覆蓋(在dispatch之前調(diào)用)總是返回false，所有Dispatcher都會(huì)調(diào)用dispatch。

在協(xié)程中，有個(gè)不成文的約定，那就是，suspend函數(shù)默認(rèn)是不阻塞線程的，也就是說，suspend函數(shù)的調(diào)用者，不用為suspend函數(shù)運(yùn)行在哪個(gè)線程而擔(dān)心，suspend函數(shù)會(huì)自己處理它工作的線程，不大部分時(shí)候，都是通過withContext來進(jìn)行切換的。

生成狀態(tài)機(jī)

免責(zé)聲明：文章其余部分所展示的代碼將不完全符合編譯器所生成的字節(jié)碼。它將是足夠準(zhǔn)確的Kotlin代碼，使你能夠理解內(nèi)部真正發(fā)生的事情。這種表示法是由Coroutines 1.3.3版本生成的，在該庫的未來版本中可能會(huì)發(fā)生變化。

Kotlin編譯器將識(shí)別函數(shù)何時(shí)可以在內(nèi)部suspend。每個(gè)suspend point都將被表示為有限狀態(tài)機(jī)中的一個(gè)狀態(tài)。這些狀態(tài)由編譯器用標(biāo)簽表示，前面示例中的suspend函數(shù)在編譯后，會(huì)產(chǎn)生類似下面的偽代碼。

fun loginUser(userId: String, password: String, completion: Continuation<Any?>) { 
  // Label 0 -> first execution 
  val user = userRemoteDataSource.logUserIn(userId, password) 
  // Label 1 -> resumes from userRemoteDataSource 
  val userDb = userLocalDataSource.logUserIn(user) 
  // Label 2 -> resumes from userLocalDataSource 
  completion.resume(userDb) 
}

為了更好地表示狀態(tài)機(jī)，編譯器將使用一個(gè)when語句來實(shí)現(xiàn)不同的狀態(tài)。

fun loginUser(userId: String, password: String, completion: Continuation<Any?>) { 
  when(label) { 
    0 -> { // Label 0 -> first execution 
        userRemoteDataSource.logUserIn(userId, password) 
    } 
    1 -> { // Label 1 -> resumes from userRemoteDataSource 
        userLocalDataSource.logUserIn(user) 
    } 
    2 -> { // Label 2 -> resumes from userLocalDataSource 
        completion.resume(userDb) 
    } 
    else -> throw IllegalStateException(...) 
  } 
}

編譯器將suspend函數(shù)編譯成帶有Continuation參數(shù)的方法叫做CPS(Continuation-Passing-Style)變換。

這段代碼是不完整的，因?yàn)椴煌臓顟B(tài)沒有辦法分享信息。編譯器會(huì)在函數(shù)中使用相同的Continuation對(duì)象來做這件事。這就是為什么Continuation的泛型是Any? 而不是原始函數(shù)的返回類型(即User)。

此外，編譯器將創(chuàng)建一個(gè)私有類，1)持有所需的數(shù)據(jù)，2)遞歸地調(diào)用loginUser函數(shù)以恢復(fù)執(zhí)行。你可以看看下面這個(gè)生成的類的近似值。

免責(zé)聲明：注釋不是由編譯器生成的。我添加它們是為了解釋它們的作用，并使跟隨代碼更容易理解。

fun loginUser(userId: String?, password: String?, completion: Continuation<Any?>) { 
  class LoginUserStateMachine( 
    // completion parameter is the callback to the function  
    // that called loginUser 
    completion: Continuation<Any?> 
  ): CoroutineImpl(completion) { 
    // Local variables of the suspend function 
    var user: User? = null 
    var userDb: UserDb? = null 
    // Common objects for all CoroutineImpls 
    var result: Any? = null 
    var label: Int = 0 
    // this function calls the loginUser again to trigger the 
    // state machine (label will be already in the next state) and 
    // result will be the result of the previous state's computation 
    override fun invokeSuspend(result: Any?) { 
      this.result = result 
      loginUser(null, null, this) 
    } 
  } 
  ... 
}

由于invokeSuspend將僅用Continuation對(duì)象的信息來再次調(diào)用loginUser，loginUser函數(shù)簽名中的其余參數(shù)都變成了空值。在這一點(diǎn)上，編譯器只需要添加如何在狀態(tài)之間轉(zhuǎn)移的信息。

它需要做的第一件事是知道1)這是函數(shù)第一次被調(diào)用，或者2)函數(shù)已經(jīng)從之前的狀態(tài)恢復(fù)。它通過檢查傳入的continuation是否是LoginUserStateMachine類型來實(shí)現(xiàn)。

fun loginUser(userId: String?, password: String?, completion: Continuation<Any?>) { 
  ... 
  val continuation = completion as? LoginUserStateMachine ?: LoginUserStateMachine(completion) 
  ... 
}

如果是第一次，它將創(chuàng)建一個(gè)新的LoginUserStateMachine實(shí)例，并將收到的完成實(shí)例作為一個(gè)參數(shù)存儲(chǔ)起來，這樣它就能記住如何恢復(fù)調(diào)用這個(gè)實(shí)例的函數(shù)。如果不是這樣，它將只是繼續(xù)執(zhí)行狀態(tài)機(jī)(suspend函數(shù))。

現(xiàn)在，讓我們看看編譯器為在狀態(tài)間移動(dòng)和在狀態(tài)間共享信息而生成的代碼。

/* Copyright 2019 Google LLC.  
   SPDX-License-Identifier: Apache-2.0 */ 
fun loginUser(userId: String?, password: String?, completion: Continuation<Any?>) { 
    ... 
 
    val continuation = completion as? LoginUserStateMachine ?: LoginUserStateMachine(completion) 
 
    when(continuation.label) { 
        0 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Next time this continuation is called, it should go to state 1 
            continuation.label = 1 
            // The continuation object is passed to logUserIn to resume  
            // this state machine's execution when it finishes 
            userRemoteDataSource.logUserIn(userId!!, password!!, continuation) 
        } 
        1 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Gets the result of the previous state 
            continuation.user = continuation.result as User 
            // Next time this continuation is called, it should go to state 2 
            continuation.label = 2 
            // The continuation object is passed to logUserIn to resume  
            // this state machine's execution when it finishes 
            userLocalDataSource.logUserIn(continuation.user, continuation) 
        } 
        ... // leaving out the last state on purpose 
    } 
}

花點(diǎn)時(shí)間瀏覽一下上面的代碼，看看你是否能發(fā)現(xiàn)與前面的代碼片斷的不同之處。讓我們看看編譯器生成了什么。

when語句的參數(shù)是LoginUserStateMachine實(shí)例中的Label。
每次處理一個(gè)新的狀態(tài)時(shí)，都會(huì)有一個(gè)檢查，以防這個(gè)函數(shù)suspend時(shí)發(fā)生異常。
在調(diào)用下一個(gè)suspend函數(shù)(即logUserIn)之前，LoginUserStateMachine實(shí)例的Label將被更新為下一個(gè)狀態(tài)。
當(dāng)在這個(gè)狀態(tài)機(jī)內(nèi)部有一個(gè)對(duì)另一個(gè)suspend函數(shù)的調(diào)用時(shí)，continuation的實(shí)例(LoginUserStateMachine類型)被作為一個(gè)參數(shù)傳遞。要調(diào)用的suspend函數(shù)也已經(jīng)被編譯器轉(zhuǎn)化了，它是另一個(gè)像這樣的狀態(tài)機(jī)，它把一個(gè)continuation對(duì)象也作為參數(shù)!當(dāng)那個(gè)suspend函數(shù)的狀態(tài)機(jī)完成后，它將恢復(fù)這個(gè)狀態(tài)機(jī)的執(zhí)行。

最后一個(gè)狀態(tài)是不同的，因?yàn)樗仨毣謴?fù)調(diào)用這個(gè)函數(shù)的執(zhí)行，正如你在代碼中看到的，它對(duì)存儲(chǔ)在LoginUserStateMachine中的cont變量(在構(gòu)造時(shí))調(diào)用resume。

/* Copyright 2019 Google LLC.  
   SPDX-License-Identifier: Apache-2.0 */ 
fun loginUser(userId: String?, password: String?, completion: Continuation<Any?>) { 
    ... 
 
    val continuation = completion as? LoginUserStateMachine ?: LoginUserStateMachine(completion) 
 
    when(continuation.label) { 
        ... 
        2 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Gets the result of the previous state 
            continuation.userDb = continuation.result as UserDb 
            // Resumes the execution of the function that called this one 
            continuation.cont.resume(continuation.userDb) 
        } 
        else -> throw IllegalStateException(...) 
    } 
}

正如你所看到的，Kotlin編譯器為我們做了很多事情!從這個(gè)suspend函數(shù)功能來舉例。

suspend fun loginUser(userId: String, password: String): User { 
  val user = userRemoteDataSource.logUserIn(userId, password) 
  val userDb = userLocalDataSource.logUserIn(user) 
  return userDb 
}

編譯器為我們生成了下面這一切。

/* Copyright 2019 Google LLC.  
   SPDX-License-Identifier: Apache-2.0 */ 
fun loginUser(userId: String?, password: String?, completion: Continuation<Any?>) { 
 
    class LoginUserStateMachine( 
        // completion parameter is the callback to the function that called loginUser 
        completion: Continuation<Any?> 
    ): CoroutineImpl(completion) { 
        // objects to store across the suspend function 
        var user: User? = null 
        var userDb: UserDb? = null 
 
        // Common objects for all CoroutineImpl 
        var result: Any? = null 
        var label: Int = 0 
 
        // this function calls the loginUser again to trigger the  
        // state machine (label will be already in the next state) and  
        // result will be the result of the previous state's computation 
        override fun invokeSuspend(result: Any?) { 
            this.result = result 
            loginUser(null, null, this) 
        } 
    } 
 
    val continuation = completion as? LoginUserStateMachine ?: LoginUserStateMachine(completion) 
 
    when(continuation.label) { 
        0 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Next time this continuation is called, it should go to state 1 
            continuation.label = 1 
            // The continuation object is passed to logUserIn to resume  
            // this state machine's execution when it finishes 
            userRemoteDataSource.logUserIn(userId!!, password!!, continuation) 
        } 
        1 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Gets the result of the previous state 
            continuation.user = continuation.result as User 
            // Next time this continuation is called, it should go to state 2 
            continuation.label = 2 
            // The continuation object is passed to logUserIn to resume  
            // this state machine's execution when it finishes 
            userLocalDataSource.logUserIn(continuation.user, continuation) 
        } 
        2 -> { 
            // Checks for failures 
            throwOnFailure(continuation.result) 
            // Gets the result of the previous state 
            continuation.userDb = continuation.result as UserDb 
            // Resumes the execution of the function that called this one 
            continuation.cont.resume(continuation.userDb) 
        } 
        else -> throw IllegalStateException(...) 
    } 
}

Kotlin編譯器將每個(gè)suspend函數(shù)轉(zhuǎn)化為一個(gè)狀態(tài)機(jī)，在每次函數(shù)需要suspend時(shí)使用回調(diào)進(jìn)行優(yōu)化。

現(xiàn)在你知道了編譯器在編譯時(shí)到底做了什么，你就可以更好地理解為什么一個(gè)suspend函數(shù)在它執(zhí)行完所有工作之前不會(huì)返回。另外，你也會(huì)知道，代碼是如何在不阻塞線程的情況下進(jìn)行suspend的——這是因?yàn)?，?dāng)函數(shù)恢復(fù)時(shí)需要執(zhí)行的信息被存儲(chǔ)在Continuation對(duì)象中!

向大家推薦下我的網(wǎng)站 https://xuyisheng.top/ 專注 Android-Kotlin-Flutter 歡迎大家訪問