Coroutine

前言

協程是一個并發方案，也是一種思想，

傳統意義上的協程是單執行緒的，面對io密集型任務他的記憶體消耗更少，進而效率高，但是面對計算密集型的任務不如多執行緒并行運算效率高，

不同的語言對于協程都有不同的實作，甚至同一種語言對于不同平臺的作業系統都有對應的實作，

我們kotlin語言的協程是 coroutines for jvm的實作方式，底層原理也是利用java 執行緒，

基礎知識

生態架構

QQ截圖20220228141259-16460288074381.png

為什么一些人總覺得協程晦澀難懂？

1.網路上沒有詳細的關于協程的概念定義，每種語言、每個系統對其實作都不一樣，可謂是眾說紛紜，什么內核態用戶態巴拉巴拉，很容易給我們帶偏

2.kotlin的各種語法糖對我們造成的干擾，如：

高階函式
原始碼實作類找不到

所以扎實的kotlin語法基本功是學習協程的前提，

實在看不懂得地方就反編譯為java，以java最終翻譯為準，

協程是什么？有什么用？

kotlin中的協程干的事就是把異步回呼代碼拍扁了，捋直了，讓異步回呼代碼同步化，除此之外，沒有任何特別之處，

創建一個協程，就是編譯器背后偷偷生成一系列代碼，比如說狀態機，

通過掛起和恢復讓狀態機狀態流轉實作把層層嵌套的回呼代碼變成像同步代碼那樣直觀、簡潔，

它不是什么執行緒框架，也不是什么高深的內核態，用戶態，它其實對于咱們安卓來說，就是一個關于回呼函式的語法糖，，，

本文將會圍繞掛起與恢復徹底剖析協程的實作原理

Kotlin函式基礎知識復習

再Kotlin中函式是一等公民，有自己的型別

函式型別

fun foo(){}
//型別為 () -> Unit
fun foo(p: Int){}
//型別為 (Int) -> String

class Foo{
    fun bar(p0: String,p1: Long):Any{}
    
}
//那么 bar 的型別為：Foo.(String,Long) -> Any
//Foo就是bar的 receiver，也可以寫成 (Foo,String,Long) ->Any

函式參考

fun foo(){} 
//參考是 ::foo
fun foo(p0: Int): String
//參考也是 ::foo

咋都一樣？沒辦法，就這樣規定的，使用的時候只能靠編譯器推斷

val f: () -> Unit = ::foo //編譯器會推斷出是fun foo(){} 
val g: (Int) -> String = ::foo //推斷為fun foo(p0: Int): String

帶Receiver的寫法

class Foo{
    fun bar(p0: String,p1: Long):Any{}
}

val h: (Foo,String,Long) -> Any = Foo:bar

系結receiver的函式參考：

val foo: Foo = Foo()
val m: (String,Long) -> Any = foo:bar

額外知識點

val x: (Foo,String,Long) -> Any = Foo:bar
val y: Function3<Foo,String,Long,Any> = x

Foo.(String,Long) -> Any = (Foo,String,Long) ->Any = Function3<Foo,String,Long,Any>

函式作為引數傳遞

fun yy(p: (Foo,String,Long)->Any){
	p(Foo(),"Hello",3L)//直接p()就能呼叫
    //p.invoke(Foo(),"Hello",3L) 也可以用invoke形式
}

Lambda

就是匿名函式，它跟普通函式比是沒有名字的，聽起來好像是廢話

//普通函式
fun func(){
   println("hello");
}
?
//去掉函式名 func,就成了匿名函式
fun(){
   println("hello");    
}
?
//可以賦值給一個變數
val func = fun(){
   println("hello");    
}
?
//匿名函式的型別
val func :()->Unit = fun(){
   println("hello");    
}
?
//Lambda運算式
val func={
   print("Hello");
}
?
//Lambda型別
val func :()->String = {
print("Hello");
"Hello" //如果是Lambda中，最后一行被當作回傳值，能省掉return，普通函式則不行
}
?
//帶引數Lambda
val f1: (Int)->Unit = {p:Int ->
print(p);
}
//可進一步簡化為
val f1 = {p:Int ->
print(p);    
}
//當只有一個引數的時候，還可以寫成
val f1: (Int)->Unit = {
   print(it);
}
?

關于函式的個人經驗總結

函式跟匿名函式看起來沒啥區別，但是反編譯為java后還是能看出點差異

如果只是用普通的函式，那么他跟普通java 函式沒啥區別，

比如 fun a() 就是對應java方法public void a(){}

但是如果通過函式參考(:: a)來用這個函式，那么他并不是直接呼叫fun a()而是重新生成一個Function0

掛起函式

suspend 修飾，

掛起函式中能呼叫任何函式，

非掛起函式只能呼叫非掛起函式，

換句話說，suspend函式只能在suspend函式中呼叫，

簡單的掛起函式展示：

//com.example.studycoroutine.chapter.CoroutineRun.kt
suspend fun suspendFun(): Int {
    return 1;
}

掛起函式特殊在哪？

public static final Object suspendFun(Continuation completion) {
    return Boxing.boxInt(1);
}

這下理解suspend為啥只能在suspend里面呼叫了吧？

想要讓道貌岸然的suspend函式干活必須要先滿足它！！！就是給它里面塞入一顆球，

然后他想呼叫其他的suspend函式，只需將球繼續塞到其它的suspend方法里面，

普通函式里沒這玩意啊，所以壓根沒法呼叫suspend函式，，，

讀到這里，想必各位會有一些疑問：

question1.這不是雞生蛋生雞的問題么？第一顆球是哪來的？
question2.為啥編譯后回傳值也變了？
question3.suspendFun 如果在協程體內被呼叫，那么他的球(completion)是誰？

標準庫給我們提供的最原始工具

public fun <T> (suspend () -> T).startCoroutine(completion: Continuation<T>) {
   createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}
?
public fun <T> (suspend () -> T).createCoroutine(completion: Continuation<T>): Continuation<Unit> =
   SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)

以一個最簡單的方式啟動一個協程，

Demo-K1

fun main() {
   val b = suspend {
       val a = hello2()
       a
  }
   b.createCoroutine(MyCompletionContinuation()).resume(Unit)
}
?
suspend fun hello2() = suspendCoroutine<Int> {
   thread{
       Thread.sleep(1000)
       it.resume(10086)
  }
}
?
class MyContinuation() : Continuation<Int> {
   override val context: CoroutineContext = CoroutineName("Co-01")
   override fun resumeWith(result: Result<Int>) {
       log("MyContinuation resumeWith 結果 = ${result.getOrNull()}")
  }
}

兩個創建協程函式區別

startCoroutine 沒有回傳值 ,而createCoroutine回傳一個Continuation,不難看出是SafeContinuation

好像看起來主要的區別就是startCoroutine直接呼叫resume(Unit)，所以不用包裝成SafeContinuation，而createCoroutine則回傳一個SafeContinuation，因為不知道將會在何時何處呼叫resume，必須保證resume只呼叫一次，所以包裝為safeContinuation

SafeContinuationd的作用是為了確保只有發生異步呼叫時才掛起

分析createCoroutineUnintercepted

//kotlin.coroutines.intrinsics.CoroutinesIntrinsicsH.kt
@SinceKotlin("1.3")
public expect fun <T> (suspend () -> T).createCoroutineUnintercepted(completion: Continuation<T>): Continuation<Unit>

先說結論

其實可以簡單的理解為kotlin層面的原語，就是回傳一個協程體，

開始分析

參考代碼Demo-K1首先b 是一個匿名函式，他肯定要被編譯為一個FunctionX，同時它還被suspend修飾所以它肯定跟普通匿名函式編譯后不一樣，

編譯后的原始碼為

public static final void main() {
     Function1 var0 = (Function1)(new Function1((Continuation)null) {
        int label;
?
        @Nullable
        public final Object invokeSuspend(@NotNull Object $result) {
           Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
           Object var10000;
           switch(this.label) {
           case 0:
              ResultKt.throwOnFailure($result);
              this.label = 1;
              var10000 = TestSampleKt.hello2(this);
              if (var10000 == var3) {
                 return var3;
              }
              break;
           case 1:
              ResultKt.throwOnFailure($result);
              var10000 = $result;
              break;
           default:
              throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
          }
?
           int a = ((Number)var10000).intValue();
           return Boxing.boxInt(a);
        }
?
        @NotNull
        public final Continuation create(@NotNull Continuation completion) {
           Intrinsics.checkParameterIsNotNull(completion, "completion");
           Function1 var2 = new <anonymous constructor>(completion);
           return var2;
        }
?
        public final Object invoke(Object var1) {
           return((<undefinedtype>)this.create((Continuation)var1)).invokeSuspend(Unit.INSTANCE);
        }
    });
     boolean var1 = false;
     Continuation var7 = ContinuationKt.createCoroutine(var0, (Continuation)(newMyContinuation()));
     Unit var8 = Unit.INSTANCE;
     boolean var2 = false;
     Companion var3 = Result.Companion;
     boolean var5 = false;
     Object var6 = Result.constructor-impl(var8);
     var7.resumeWith(var6);
  }

我們可以看到先是 Function1 var0 = new Function1創建了一個物件，此時跟協程沒關系，這步只是編譯器層面的匿名函式語法優化

如果直接

fun main() {
   suspend {
       val a = hello2()
       a
  }.createCoroutine(MyContinuation()).resume(Unit)
}

也是一樣會創建Function1 var0 = new Function1

解答question1

繼續呼叫createCoroutine

再繼續createCoroutineUnintercepted ，找到在JVM平臺的實作

//kotlin.coroutines.intrinsics.IntrinsicsJVM.class
@SinceKotlin("1.3")
public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
   completion: Continuation<T>
): Continuation<Unit> {
//probeCompletion還是我們傳入completion物件，在我們的Demo就是myCoroutine
   val probeCompletion = probeCoroutineCreated(completion)//probeCoroutineCreated方法點進去看了，好像是debug用的.我的理解是這樣的
   //This就是這個suspend lambda，在Demo中就是myCoroutineFun
   return if (this is BaseContinuationImpl)
       create(probeCompletion)
   else
//else分支在我們demo中不會走到
     //當 [createCoroutineUnintercepted] 遇到不繼承 BaseContinuationImpl 的掛起 lambda 時，將使用此函式，
       createCoroutineFromSuspendFunction(probeCompletion) {
          (this as Function1<Continuation<T>, Any?>).invoke(it)
      }
}

@NotNull
public final Continuation create(@NotNull Continuation completion) {
Intrinsics.checkNotNullParameter(completion, "completion");
Function1 var2 = new <anonymous constructor>(completion);
return var2;
}

把completion傳入，并創建一個新的Function1，作為Continuation回傳，這就是創建出來的協程體物件，協程的作業核心就是它內部的狀態機，invokeSuspend函式

呼叫 create

@NotNull
public final Continuation create(@NotNull Continuation completion) {
	Intrinsics.checkNotNullParameter(completion, "completion");
	Function1 var2 = new <anonymous constructor>(completion);
	return var2;
}

補充---相關類繼承關系

解答question2&3

已知協程啟動會呼叫協程體的resume,該呼叫最侄訓來到BaseContinuationImpl::resumeWith

internal abstract class BaseContinuationImpl{
   fun resumeWith(result: Result<Any?>) {
          // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
       var current = this
       var param = result
       while (true) {
           with(current) {
               val completion = completion!! // fail fast when trying to resume continuation without completion
               val outcome: Result<Any?> =
                   try {
                       val outcome = invokeSuspend(param)//呼叫狀態機
                       if (outcome === COROUTINE_SUSPENDED) return
                       Result.success(outcome)
                  } catch (exception: Throwable) {
                       Result.failure(exception)
                  }
               releaseIntercepted() // this state machine instance is terminating
               if (completion is BaseContinuationImpl) {
                   // unrolling recursion via loop
                   current = completion
                   param = outcome
              } else {
                   //最終走到這里，這個completion就是被塞的第一顆球，
                   completion.resumeWith(outcome)
                   return
              }
          }
      }
  }
}

狀態機代碼截取

public final Object invokeSuspend(@NotNull Object $result) {
   Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
   Object var10000;
   switch(this.label) {
   case 0://第一次進來 label = 0 
      ResultKt.throwOnFailure($result);
      // label改成1了，意味著下一次被恢復的時候會走case 1，這就是所謂的【狀態流轉】
      this.label = 1; 
      //全體目光向我看齊，我宣布個事：this is 協程體物件，
      var10000 = TestSampleKt.hello2(this);
      if (var10000 == var3) {
         return var3;
      }
      break;
   case 1:
      ResultKt.throwOnFailure($result);
      var10000 = $result;
      break;
   default:
      throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
   }

   int a = ((Number)var10000).intValue();
   return Boxing.boxInt(a);
}

question3答案出來了，傳進去的是create創建的那個continuation

最后再來聊聊question2，從上面的代碼已經很清楚的告訴我們為啥掛起函式反編譯后的回傳值變為object了，

以hello2為例子，hello2能回傳代表掛起的白板，也能回傳result，如果回傳白板，狀態機return，協程掛起，如果回傳result，那么hello2執行完畢，是一個沒有掛起的掛起函式，通常編譯器也會提醒 suspend 修飾詞無意義，所以這就是設計需要，沒有啥因為所以，

最后，除了直接回傳結果的情況，掛起函式一定會以resume結尾，要么回傳result，要么回傳例外，代表這個掛起函式回傳了，

呼叫resume意義在于重新回呼BaseContinuationImpl的resumeWith，進而喚醒狀態機，繼續執行協程體的代碼，

換句話說，我們自定義的suspend函式，一定要利用suspendCoroutine 獲得續體，即狀態機物件，否則無法實作真正的掛起與resume，

suspendCoroutine

我們可以不用suspendCoroutine，用更直接的suspendCoroutineUninterceptedOrReturn也能實作，不過這種方式要手動回傳白板，不過一定要小心，要在合理的情況下回傳或者不回傳，不然會產生很多意想不到的結果

suspend fun mySuspendOne() = suspendCoroutineUninterceptedOrReturn<String> { continuation ->
    thread {
        TimeUnit.SECONDS.sleep(1)
        continuation.resume("hello world")
    }
    //因為我們這個函式沒有回傳正確結果，所以必須回傳一個掛起標識,否則BaseContinuationImpl會認為完成了任務， 
    // 并且我們的執行緒又在運行沒有取消，這將很多意想不到的結果
    kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
}

而suspendCoroutine則沒有這個隱患

suspend fun mySafeSuspendOne() = suspendCoroutine<String> { continuation ->
    thread {
        TimeUnit.SECONDS.sleep(1)
        continuation.resume("hello world")
    }
    //suspendCoroutine函式很聰明的幫我們判斷回傳結果如果不是想要的物件，自動返				
    kotlin.coroutines.intrinsics.COROUTINE_SUSPENDED
}

public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T =
    suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
    	//封裝一個代理Continuation物件
        val safe = SafeContinuation(c.intercepted())
        block(safe)
        //根據block回傳結果判斷要不要回傳COROUTINE_SUSPENDED
        safe.getOrThrow()
    }

SafeContinuation的奧秘

//呼叫單引數的這個構造方法
internal actual constructor(delegate: Continuation<T>) : this(delegate, UNDECIDED)
@Volatile
private var result: Any? = initialResult //UNDECIDED賦值給 result 
//java原子屬性更新器那一套東西
private companion object {
       @Suppress("UNCHECKED_CAST")
       @JvmStatic
       private val RESULT = AtomicReferenceFieldUpdater.newUpdater<SafeContinuation<*>, Any?>(
           SafeContinuation::class.java, Any::class.java as Class<Any?>, "result"
      )
  }
?
internal actual fun getOrThrow(): Any? {
   var result = this.result // atomic read
   if (result === UNDECIDED) { //如果UNDECIDED，那么就把result設定為COROUTINE_SUSPENDED
       if (RESULT.compareAndSet(this, UNDECIDED, COROUTINE_SUSPENDED)) returnCOROUTINE_SUSPENDED
       result = this.result // reread volatile var
  }
   return when {
       result === RESUMED -> COROUTINE_SUSPENDED // already called continuation, indicate COROUTINE_SUSPENDED upstream
       result is Result.Failure -> throw result.exception
       else -> result // either COROUTINE_SUSPENDED or data <-這里回傳白板
  }
}
?
public actual override fun resumeWith(result: Result<T>) {
       while (true) { // lock-free loop
           val cur = this.result // atomic read，不理解這里的官方注釋為啥叫做原子讀，我覺得 Volatile只能保證可見性，
           when {
             //這里如果是UNDECIDED 就把 結果附上去，
               cur === UNDECIDED -> if (RESULT.compareAndSet(this, UNDECIDED, result.value)) return
             //如果是掛起狀態，就通過resumeWith回呼狀態機
               cur === COROUTINE_SUSPENDED -> if (RESULT.compareAndSet(this, COROUTINE_SUSPENDED, RESUMED)){
                   delegate.resumeWith(result)
                   return
              }
               else -> throw IllegalStateException("Already resumed")
          }
      }
  }

val safe = SafeContinuation(c.intercepted())
block(safe)
safe.getOrThrow()

先回顧一下什么叫真正的掛起，就是getOrThrow回傳了“白板”，那么什么時候getOrThrow能回傳白板？答案就是result被初始化后值沒被修改過，那么也就是說resumeWith沒有被執行過，即：block(safe)這句代碼，block這個被傳進來的函式，執行程序中沒有呼叫safe的resumeWith，原理就是這么簡單，cas代碼保證關鍵邏輯的原子性與并發安全

繼續以Demo-K1為例子，這里假設hello2運行在一條新的子執行緒，否則仍然是沒有掛起，

{
   thread{
       Thread.sleep(1000)
       it.resume(10086)
  }
}

總結

最后，可以說開啟一個協程，就是利用編譯器生成一個狀態機物件，幫我們把回呼代碼拍扁，成為同步代碼，

轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/452072.html

標籤：Android

上一篇：拿捏了！這樣的購物體驗才深得用戶心

下一篇：iOS全埋點解決方案-應用退出和啟動

Android中的Coroutine協程原理詳解

前言