对于suspend修饰的函数，编译器会帮我们在该函数中传入一个Continuation参数，使用Continuation参数代替了suspend修饰符。

  1. public interface Continuation<in T> {
  2.     // 与该续体对应的协程的上下文
  3.     public val context: CoroutineContext
  4.     // 恢复对应协程的执行，并且传递一个表示成功或失败的result作为最后一个挂起点的返回值
  5.     public fun resumeWith(result: Result<T>)
  6. }

在协程中使用Continuation接口表示一个续体，它代表一个挂起点之后的延续，即 挂起点之后的剩余应执行的代码。
（比如：进行网络请求的时候，使用Callback接收网络请求的结果，这时候可以将Callback看作一个续体，即网络请求的续体，用于接收网络请求的结果）

通过传递Continuation来控制异步调用流程被称作CPS变换（Continuation-Passing-Style Transformation）。

  1. //正常的代码
  3. class Temp {
  4.     suspend fun fetchData(argument: String): Boolean {
  5.         val result = netRequest(argument)
  6.         return result == 0
  7.     }
  9.     // 模拟网络请求
  10.     suspend fun netRequest(argument: String): Int {
  11.         delay(1000)
  12.         return argument.length
  13.     }
  14. }

fetchData函数编译时生成了一个静态内部类

  1. //反编译后的简略版代码
  2. static final class FetchDataStateMachine extends ContinuationImpl {
  3.     Object result;
  4.     int label;
  6.     FetchDataStateMachine(Continuation $completion) {
  7.         super($completion);  //接收一个名称为$completion的Continuation参数，$completion被保存在父类BaseContinuationImpl中：
  8.     }
  10.     @Nullable
  11.     public final Object invokeSuspend(@NotNull Object $result) {
  12.         this.result = $result;
  13.         this.label |= Integer.MIN_VALUE;
  14.         return Temp.this.fetchData(null, (Continuation<? super Boolean>) this);
  15.     }
  16. }
  1. //继承关系
  2. FetchDataStateMachine -> ContinuationImpl -> BaseContinuationImpl -> Continuation
  3. //
  4. 所以，也可以说：fetchData函数编译时生成了续体。
  1. /*
  2. BaseContinuationImpl主要做了下面的几件事情：
  4.    保存completion：它保存了fetchData方法的FetchDataStateMachine实例，使得可以一级一级地向上回调续体。
  6.   （重写resumeWith方法：BaseContinuationImpl重写了Continuation接口的resumeWith方法，该方法用于恢复协程，也是协程恢复的核心逻辑）。
  7. */
  10. internal abstract class BaseContinuationImpl(
  11.     public val completion: Continuation<Any?>?
  12. ) : Continuation<Any?>, CoroutineStackFrame, Serializable {
  14.     public final override fun resumeWith(result: Result<Any?>) {
  15.         var current = this
  16.         var param = result  
  17.         while (true) { 
  18.             probeCoroutineResumed(current)
  19.             with(current) {
  20.                 val completion = completion!! 
  21.                 val outcome: Result<Any?> =
  22.                     try {
  23.                         val outcome = invokeSuspend(param)  
  24.                         if (outcome === COROUTINE_SUSPENDED) return
  25.                         Result.success(outcome)
  26.                     } catch (exception: Throwable) {
  27.                         Result.failure(exception)
  28.                     }
  29.                 releaseIntercepted() 
  30.                 if (completion is BaseContinuationImpl) {
  32.                     current = completion
  33.                     param = outcome
  34.                 } else {
  36.                     completion.resumeWith(outcome)
  37.                     return
  38.                 }
  39.             }
  40.         }
  41.     }

状态机FetchDataStateMachine声明了result和label两个变量

  • result：表示上一个Continuation的结果，比如有函数A和B，函数内部分别声明了ContinuationA和ContinuationB，A调用B并且将ContinuationA传入B中保存。在后续回调的过程中，ContinuationA可以从result变量中拿到ContinuationB::invokeSuspend的执行结果。

  • label：Kotlin Compiler可以识别函数内部哪个地方会挂起，每一个挂起点（suspension point）被表示为状态机的一个状态（state），这些状态通过switch case语句表示出来。label表示当前应该执行状态机的哪一个状态，具体来说就是要进入哪一个case，通过label变量就记录下了状态机当前的状态。

  1. //反编译后的代码
  3. public final Object fetchData(@NotNull String argument,@NotNull Continuation completion) {
  4.     Object $continuation;
  5.     label25:
  6.     {    
  7.         /*
  8.         判断传入的completion是否为FetchDataStateMachine类型，
  9.         若是则对它的label变量做些操作，
  10.         若不是则直接创建一个FetchDataStateMachine并且传入completion（completion会被保存下来）。
  11.         */
  12.         if (completion instanceof FetchDataStateMachine) {
  13.             $continuation = (FetchDataStateMachine) completion;
  14.             if (($continuation.label & Integer.MIN_VALUE) != 0) {
  15.                 $continuation.label -= Integer.MIN_VALUE;
  16.                 break label25;
  17.             }
  18.         }
  19.         $continuation = new FetchDataStateMachine(completion);
  20.     }
  23.     /*
  24.     fetchData方法原先的代码语句会被划分为switch下的多个case语句
  25.     label变量控制当前要执行哪个case分支。
  26.     */
  27.     Object $result = $continuation.result;
  28.     Object resultTemp;
  29.     switch ($continuation.label) {
  30.         case 0:
  31.             ResultKt.throwOnFailure($result);
  32.             $continuation.label = 1;
  33.             resultTemp = this.netRequest(argument, (Continuation) $continuation);
  34.             if (resultTemp == COROUTINE_SUSPENDED) {
  35.                 return COROUTINE_SUSPENDED;
  36.             }
  37.             break;
  38.         case 1:
  39.             ResultKt.throwOnFailure($result);
  40.             resultTemp = $result;
  41.             break;
  42.         default:
  43.             throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
  44.     }
  46.     int result = ((Number) resultTemp).intValue();
  47.     return Boxing.boxBoolean(result == 0);
  48. }
  50. static final class FetchDataStateMachine extends ContinuationImpl {
  51.     Object result;
  52.     int label;
  54.     FetchDataStateMachine(Continuation $completion) {
  55.         super($completion);
  56.     }
  58.     @Nullable
  59.     public final Object invokeSuspend(@NotNull Object $result) {
  60.         this.result = $result;
  61.         this.label |= Integer.MIN_VALUE;
  62.         return Temp.this.fetchData(null, (Continuation<? super Boolean>) this);
  63.     }
  64. }

有限状态机（FSM，即 Finite-State Machine）——来控制协程代码的执行。

  1. // 模拟网络请求
  2.     suspend fun netRequest(argument: String): Int {
  3.         delay(1000)
  4.         return argument.length
  5.     }
  10. //反编译后
  11. public static final Object delay(long timeMillis, @NotNull Continuation $completion) {
  12.     if (timeMillis <= 0L) {
  13.         return Unit.INSTANCE;
  14.     } else {
  15.         // 实现类
  16.         CancellableContinuationImpl cancellableContinuationImpl = new CancellableContinuationImpl(IntrinsicsKt.intercepted($completion), 1);
  17.         cancellableContinuationImpl.initCancellability();
  18.         // 向上转型
  19.         CancellableContinuation cont = (CancellableContinuation)cancellableContinuationImpl;
  20.         if (timeMillis < Long.MAX_VALUE) {
  21.             // 延时操作
  22.             getDelay(cont.getContext()).scheduleResumeAfterDelay(timeMillis, cont);
  23.         }
  24.         // 获取执行结果
  25.         Object result = cancellableContinuationImpl.getResult();
  26.         if (result == COROUTINE_SUSPENDED) {
  27.             DebugProbesKt.probeCoroutineSuspended($completion);
  28.         }
  29.         // 返回结果
  30.         return result;
  31.     }
  32. }

当delay方法需要挂起的时候，它返回COROUTINE_SUSPENDED，接着netRequest方法返回COROUTINE_SUSPENDED，接着fetchData方法返回COROUTINE_SUSPENDED，重复这个过程直到调用栈的最上层。
通过这种「结束方法调用」的方式，让协程暂时不在这个线程上面执行，让线程可以去处理其它的任务（包括执行其它的协程），这也就是为什么协程的挂起不会阻塞当前的线程，也是「非阻塞式挂起」。

恢复

  1. // 模拟网络请求
  2.     suspend fun netRequest(argument: String): Int {
  3.         delay(1000)
  4.         return argument.length
  5.     }
  8. // 反编译后
  10. public final Object netRequest(@NotNull String argument,@NotNull Continuation completion) {
  11.     Object $continuation;
  12.     label20:
  13.     {
  14.         if (completion instanceof NetRequestStateMachine) {
  15.             $continuation = (NetRequestStateMachine) completion;
  16.             if (($continuation.label & Integer.MIN_VALUE) != 0) {
  17.                 $continuation.label -= Integer.MIN_VALUE;
  18.                 break label20;
  19.             }
  20.         }
  21.         $continuation = new NetRequestStateMachine(completion);
  22.     }
  24.     Object $result = $continuation.result;
  25.     switch ($continuation.label) {
  26.         case 0:
  27.             ResultKt.throwOnFailure($result);
  28.             $continuation.functionParameter = argument;
  29.             $continuation.label = 1;
  30.             if (DelayKt.delay(1000L, (Continuation) $continuation) == COROUTINE_SUSPENDED) {
  31.                 return COROUTINE_SUSPENDED;
  32.             }
  33.             break;
  34.         case 1:
  35.             argument = (String) ($continuation.functionParameter);
  36.             ResultKt.throwOnFailure($result);
  37.             break;
  38.         default:
  39.             throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
  40.     }
  42.     return Boxing.boxInt(argument.length());
  43. }
  45. static final class NetRequestStateMachine extends ContinuationImpl {
  46.     Object result;
  47.     int label;
  48.     Object functionParameter;
  50.     NetRequestStateMachine(Continuation $completion) {
  51.         super($completion);
  52.     }
  54.     @Nullable
  55.     public final Object invokeSuspend(@NotNull Object $result) {         // invokeSuspend
  56.         this.result = $result;
  57.         this.label |= Integer.MIN_VALUE;
  58.         return Temp.this.netRequest(null, (Continuation<? super Integer>) this);
  59.     }
  60. }

resumeWith方法 和 恢复 相关

  1. internal abstract class BaseContinuationImpl(
  2.     public val completion: Continuation<Any?>?
  3. ) : Continuation<Any?>, CoroutineStackFrame, Serializable {
  5.     // This implementation is final. This fact is used to unroll resumeWith recursion.
  6.     public final override fun resumeWith(result: Result<Any?>) {
  7.         // This loop unrolls recursion in current.resumeWith(param) to make saner and shorter stack traces on resume
  8.         var current = this
  9.         var param = result
  10.         while (true) {
  11.             // 在每个恢复的continuation进行调试探测，使得调试库可以精确跟踪挂起的调用栈中哪些部分
  12.             // 已经恢复了。
  13.             probeCoroutineResumed(current)
  14.             with(current) {
  15.                 val completion = completion!! // fail fast when trying to resume continuation without completion
  16.                 val outcome: Result<Any?> =
  17.                     try {
  19.                         val outcome = invokeSuspend(param)
  20.                         if (outcome === COROUTINE_SUSPENDED) return
  21.                         Result.success(outcome)
  22.                     } catch (exception: Throwable) {
  23.                         Result.failure(exception)
  24.                     }
  25.                 releaseIntercepted() // this state machine instance is terminating
  26.                 if (completion is BaseContinuationImpl) { // 符合
  27.                      //把current更新为FetchDataStateMachine实例
  28.                     current = completion
  29.                         // 把param更新为outcome(包装了netRequest-Return的Result)
  30.                     param = outcome
  31.                 } else {
  32.                     // top-level completion reached -- invoke and return
  33.                     completion.resumeWith(outcome)
  34.                     return
  35.                 }
  36.             }
  37.         }
  38.     }
  40.     protected abstract fun invokeSuspend(result: Result<Any?>): Any?
  42.     protected open fun releaseIntercepted() {
  43.         // does nothing here, overridden in ContinuationImpl
  44.     }    
  46.     ...
  47. }
  1. static final class NetRequestStateMachine extends ContinuationImpl {
  3.  .......
  4.     @Nullable
  5.     public final Object invokeSuspend(@NotNull Object $result) {
  6.         this.result = $result;
  7.         this.label |= Integer.MIN_VALUE;
  8.      /*
  9.      调用了netRequest方法，并且将续体自身作为参数传入。
  10.      */
  11.         return Temp.this.netRequest(null, (Continuation<? super Integer>) this);
  12.      /*
  13.      该返回值会传递到BaseContinuationImpl的resumeWith方法中
  14.      */
  15.     }
  16. }
  1. NetRequestStateMachine -> ContinuationImpl -> BaseContinuationImpl -> Continuation
  1. BaseContinuationImpl(
  3.     resumeWith(result: Result<Any?>)
  5.         val outcome: Result<Any?> =
  6.                     try {
  7.                         val outcome = invokeSuspend(param)
  8.                         if (outcome === COROUTINE_SUSPENDED) return
  9.                         Result.success(outcome)
  10.                     } catch (exception: Throwable) {
  11.                         Result.failure(exception)
  12.                     }
  15. 在resumeWith方法中，将netRequest-Return包装为Result保存到outcome变量中。

【续体的调用过程，其实就是层层往上地调用续体的invokeSuspend方法】
继续进行下一轮while循环，在with块中会执行FetchDataStateMachine::invokeSuspend

要知道怎么恢复——就要知道调用fetchData方法的时候会做些什么。

  2.  // 挂起函数
  3. suspend fun requestToken(): Token { ... }  
  5. //launch方法可以启动一个协程
  6. fun postItem(item: Item) {
  7.     GlobalScope.launch {
  8.         val token = requestToken() // 
  9.         val post = createPost(token, item)
  10.         processPost(post)
  11.     }
  12. }
  14. //源码---------------------------------
  16. public fun CoroutineScope.launch(
  17.     context: CoroutineContext = EmptyCoroutineContext,
  18.     start: CoroutineStart = CoroutineStart.DEFAULT,
  19.     block: suspend CoroutineScope.() -> Unit
  20. ): Job {
  21.     ...
  22.     coroutine.start(start, coroutine, block) 
  23.     return coroutine
  24. }

coroutine.start(start, coroutine, block)
|
走到｜
｜
startCoroutineCancellable

  1. internal fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>) =
  2.     createCoroutineUnintercepted(completion).intercepted().resumeCancellable(Unit)
  5. public fun <T> Continuation<T>.resumeCancellableWith(
  6.     result: Result<T>,
  7.     onCancellation: ((cause: Throwable) -> Unit)? = null
  8. ): Unit = when (this) {  // 判断
  9.     is DispatchedContinuation -> resumeCancellableWith(result, onCancellation)
  10.     else -> resumeWith(result) // 传递
  11. }
  14. /*
  15. 首先会判断当前续体类型是否是DispatchedContinuation（有没有被调度器拦截包装），
  16. 如果没有被拦截直接调用resumeWith()恢复协程代码块的执行，
  17. 否则调用拦截包装后的DispatchedContinuation类型的resumeCancellableWith():
  18. */
  1. //包装后的续体类型，维护了调度器dispatcher和原续体对象continuation
  3. internal class DispatchedContinuation<in T>(
  4.     @JvmField val dispatcher: CoroutineDispatcher,
  5.     @JvmField val continuation: Continuation<T>
  6. ) : DispatchedTask<T>(MODE_UNINITIALIZED), CoroutineStackFrame, Continuation<T> by continuation {
  8.     //DispatchedContinuation类的resumeCancellableWith()
  9.     inline fun resumeCancellableWith(
  10.         result: Result<T>,
  11.         noinline onCancellation: ((cause: Throwable) -> Unit)?
  12.     ) {
  13.         val state = result.toState(onCancellation)
  14.         //通过调度器判断是否需要切换线程
  15.         if (dispatcher.isDispatchNeeded(context)) {
  16.             _state = state
  17.             resumeMode = MODE_CANCELLABLE
  18.             //A 如果需要，则切换线程，将当前续体this作为Runnable传入
  19.             dispatcher.dispatch(context, this)
  20.         } else {
  21.             executeUnconfined(state, MODE_CANCELLABLE) {
  22.                 if (!resumeCancelled(state)) {
  23.                     //B 该函数最终会调用原continuation.resumeWith(result)直接在当前线程恢复协程执行
  24.                     resumeUndispatchedWith(result) 
  25.                 }
  26.             }
  27.         }
  28.     }
  30.     //B 没有调度器拦截的情况直接在当前线程执行
  31.     inline fun resumeUndispatchedWith(result: Result<T>) {
  32.         withCoroutineContext(context, countOrElement) {
  33.             continuation.resumeWith(result)  
  34.         }
  35.     }
  36. }

协程恢复的实质是对续体进行回调。

kotlin 协程原理 - 图1