前言
協程是一個併發方案。也是一種思想。
傳統意義上的協程是單線程的，面對io密集型任務他的內存消耗更少，進而效率高。但是面對計算密集型的任務不如多線程並行運算效率高。
不同的語言對於協程都有不同的實現，甚至同一種語言對於不同平台的操作系統都有對應的實現。
我們kotlin語言的協程是 coroutines for jvm的實現方式。底層原理也是利用java 線程。
基礎知識
生態架構
相關依賴庫
dependencies {
// Kotlin
implementation "org.jetbrains.kotlin:kotlin-stdlib:1.4.32"
// 協程核心庫
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.4.3"
// 協程Android支持庫
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.4.3"
// 協程Java8支持庫
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-jdk8:1.4.3"
// lifecycle對於協程的擴展封裝
implementation "androidx.lifecycle:lifecycle-viewmodel-ktx:2.2.0"
implementation "androidx.lifecycle:lifecycle-runtime-ktx:2.2.0"
implementation "androidx.lifecycle:lifecycle-livedata-ktx:2.2.0"
}
為什麼一些人總覺得協程晦澀難懂？
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;
}
把completion傳入，並創建一個新的Function1，作為Continuation返回，這就是創建出來的協程體對象，協程的工作核心就是它內部的狀態機，invokeSuspend函數
補充---相關類繼承關係
解答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)
}
}
總結
最後，可以説開啓一個協程，就是利用編譯器生成一個狀態機對象，幫我們把回調代碼拍扁，成為同步代碼。
