Kotlin笔记之协程工作原理

也想出现在这里？点击联系我~

协程的状态机

这一章会以下面的代码为例解析一下协程启动，挂起以及恢复的流程：

1. privatesuspendfungetId():String{
2. returnGlobalScope.async(Dispatchers.IO){
3. delay(1000)
4. "hearing"
5. }.await()
6. }
8. privatesuspendfungetAvatar(id:String):String{
9. returnGlobalScope.async(Dispatchers.IO){
10. delay(1000)
11. "avatar-$id"
12. }.await()
13. }
15. funmain(){
16. GlobalScope.launch{
17. valid=getId()
18. valavatar=getAvatar(id)
19. println("${Thread.currentThread().name}-$id-$avatar")
20. }
21. }

上面 main 方法中，GlobalScope.launch 启动的协程体在执行到 getId 后，协程体会挂起，直到 getId 返回可用结果，才会 resume launch 协程，执行到 getAvatar 也是同样的过程。协程内部实现使用状态机来处理不同的挂起点，将 GlobalScope.launch 协程体字节码反编译成 Java 代码，大致如下(有所删减)：

1. BuildersKt.launch$default((CoroutineScope)GlobalScope.INSTANCE,(CoroutineContext)null,
2. (CoroutineStart)null,(Function2)(newFunction2((Continuation)null){
3. intlabel;
5. publicfinalObjectinvokeSuspend(
7. Object$result){
8. Objectvar10000;
9. Stringid;
10. label17:{
11. CoroutineScope$this$launch;
12. switch(this.label){
13. case0://a
14. ResultKt.throwOnFailure($result);
15. $this$launch=this.p$;
16. this.label=1;//label置为1
17. var10000=getId(this);
18. if(var10000==COROUTINE_SUSPENDED){
19. returnCOROUTINE_SUSPENDED;
20. }
21. //若此时已经有结果，则不挂起，直接break
22. break;
23. case1://b
24. ResultKt.throwOnFailure($result);
25. var10000=$result;
26. break;
27. case2://d
28. id=(String)this.L$1;
29. ResultKt.throwOnFailure($result);
30. var10000=$result;
31. breaklabel17;//退出label17
32. default:
33. thrownewIllegalStateException("callto'resume'before'invoke'withcoroutine");
34. }
35. //c
36. id=(String)var10000;
37. this.L$1=id;//将id赋给L$1
38. this.label=2;//label置为2
39. var10000=getAvatar(id,this);
40. if(var10000==COROUTINE_SUSPENDED){
41. returnCOROUTINE_SUSPENDED;
42. }
43. }
44. //e
45. Stringavatar=(String)var10000;
46. Stringvar5=var9.append(var10001.getName()).append("-").append(id).append("-").append(avatar).toString();
47. System.out.println(var5);
48. returnUnit.INSTANCE;
49. }
54. publicfinalContinuationcreate(
56. Objectvalue,
58. Continuationcompletion){
59. Intrinsics.checkParameterIsNotNull(completion,"completion");
60. Function2var3=new<anonymousconstructor>(completion);
61. var3.p$=(CoroutineScope)value;
62. returnvar3;
63. }
65. publicfinalObjectinvoke(Objectvar1,Objectvar2){
66. return((<undefinedtype>)this.create(var1,(Continuation)var2)).invokeSuspend(Unit.INSTANCE);
67. }
68. }

这里我们根据上面的注释以及字母标签来看一下执行流程(invokeSuspend 方法会在协程体中的 suspend 函数得到结果后被调用，具体是在哪里被调用的稍后会讲到)：

• a: launch 协程体刚执行到 getId 方法时，getId 方法的返回值将是 COROUTINE_SUSPENDED, 此时直接 return, 则 launch 协程体中 getId 后面的代码暂时不会执行，即 launch 协程体被挂起(非阻塞, 该线程依旧会做其它工作)。这里将 label 置为了 1. 而若此时 getId 已经有结果(内部没有调用 delay 之类的 suspend 函数等)，则不挂起，而是直接 break。
• b: 若上面 a 中 getId 返回 COROUTINE_SUSPENDED, 则当 getId 有可用结果返回后，会重新执行 launch 协程体的 invokeSuspend 方法，根据上面的 label==1, 会执行到这里检查一下 result 没问题的话就 break, 此时 id 赋值给了 var10000。
• c: 在 a 中若直接 break 或 在 b 中得到 getId 的结果然后 break 后，都会执行到这里，得到 id 的值并把 label 置为2。然后调用 getAvatar 方法，跟 getId 类似，若其返回 COROUTINE_SUSPENDED 则 return，协程被挂起，等到下次 invokeSuspend 被执行，否则离开 label17 接着执行后续逻辑。
• d: 若上面 c 中 getAvatar 返回 COROUTINE_SUSPENDED, 则当 getAvatar 有可用结果返回后会重新调用 launch 协程体的 invokeSuspend 方法，此时根据 label==2 来到这里并取得之前的 id 值，检验 result(即avatar)，然后break label17。
• e: c 中直接返回了可用结果 或 d 中 break label17 后，launch 协程体中的 suspend 函数都执行完毕了，这里会执行剩下的逻辑。

suspend 函数不会阻塞线程，且 suspend 函数不一定会挂起协程，如果相关调用的结果已经可用，则继续运行而不挂起，例如 async{} 返回值 Deferred 的结果已经可用时，await()挂起函数可以直接返回结果，不用再挂起协程。

这一节看了一下 launch 协程体反编译成 Java 后的代码逻辑，关于 invokeSuspend 是何时怎么被调用的，将会在下面讲到。

协程的创建与启动

这一节以 CoroutineScope.launch {} 默认参数为例，从源码角度看看 Kotlin 协程是怎样创建与启动的：

1. publicfunCoroutineScope.launch(
2. context:CoroutineContext=EmptyCoroutineContext,
3. start:CoroutineStart=CoroutineStart.DEFAULT,
4. block:suspendCoroutineScope.()->Unit
5. ):Job{
6. valnewContext=newCoroutineContext(context)
7. valcoroutine=if(start.isLazy)LazyStandaloneCoroutine(newContext,block)elseStandaloneCoroutine(newContext,active=true)
8. coroutine.start(start,coroutine,block)
9. returncoroutine
10. }
12. //AbstractCoroutine.kt
13. //receiver:StandaloneCoroutine
14. //block:suspendStandaloneCoroutine.()->Unit
15. //privateopenclassStandaloneCoroutine(…):AbstractCoroutine<Unit>(…){}
16. //publicabstractclassAbstractCoroutine<inT>(…):JobSupport(active),Job,Continuation<T>,CoroutineScope{}
17. publicfun<R>start(start:CoroutineStart,receiver:R,block:suspendR.()->T){
18. //调用CoroutineStart中的invoke方法
19. start(block,receiver,this)
20. }
22. publicenumclassCoroutineStart{
23. //block-StandaloneCoroutine.()->Unit
24. //receiver-StandaloneCoroutine
25. //completion-StandaloneCoroutine<Unit>
26. publicoperatorfun<R,T>invoke(block:suspendR.()->T,receiver:R,completion:Continuation<T>):Unit=
27. when(this){
28. //根据start参数的类型调用不同的方法
29. DEFAULT->block.startCoroutineCancellable(receiver,completion)
30. ATOMIC->block.startCoroutine(receiver,completion)
31. UNDISPATCHED->block.startCoroutineUndispatched(receiver,completion)
32. LAZY->Unit//willstartlazily
33. }
34. }

接下来看看 startCoroutineCancellable 方法：

1. //receiver-StandaloneCoroutine
2. //completion-StandaloneCoroutine<Unit>
3. internalfun<R,T>(suspend(R)->T).startCoroutineCancellable(receiver:R,completion:Continuation<T>)=
4. runSafely(completion){
5. createCoroutineUnintercepted(receiver,completion).intercepted().resumeCancellableWith(Result.success(Unit))
6. }

createCoroutineUnintercepted 方法创建了一个 Continuation 类型(协程)的实例，即创建了一个协程：

1. publicactualfun<R,T>(suspendR.()->T).createCoroutineUnintercepted(
2. receiver:R,completion:Continuation<T>
3. ):Continuation<Unit>{
4. returnif(thisisBaseContinuationImpl)create(receiver,completion)else//…
5. }

调用的是 (suspend (R) -> T) 的 createCoroutineUnintercepted 方法，(suspend (R) -> T) 就是协程体。直接看上面示例代码中 GlobalScope.launch 编译后的字节码，可以发现 CoroutineScope.launch 传入的 lambda 表达式被编译成了继承 SuspendLambda 的子类：

1. finalclassMain$main$1extendskotlin/coroutines/jvm/internal/SuspendLambdaimplementskotlin/jvm/functions/Function2

其继承关系为: SuspendLambda -> ContinuationImpl -> BaseContinuationImpl -> Continuation, 因此走 create(receiver, completion) 方法，从上面反编译出的 Java 代码可以看到 create 方法创建了一个 Continuation 实例，再看一下 Kotlin 代码编译后的字节码(包名已省略)：

1. publicfinalcreate(Ljava/lang/Object;Lkotlin/coroutines/Continuation;)Lkotlin/coroutines/Continuation;
2. //…
3. NEWMain$main$1

从上面可以看到，create 方法创建了 Main$main$1 实例，而其继承自 SuspendLambda, 因此 create 方法创建的 Continuation 是一个 SuspendLambda 对象。

即 createCoroutineUnintercepted 方法创建了一个 SuspendLambda 实例。然后看看 intercepted 方法：

1. publicactualfun<T>Continuation<T>.intercepted():Continuation<T>=
2. //如果是ContinuationImpl类型，则调用intercepted方法，否则返回自身
3. //这里的this是Main$main$1实例-ContinuationImpl的子类
4. (thisas?ContinuationImpl)?.intercepted()?:this
6. //ContinuationImpl
7. publicfunintercepted():Continuation<Any?>=
8. //context[ContinuationInterceptor]是CoroutineDispatcher实例
9. //需要线程调度-返回DispatchedContinuation，其continuation参数值为SuspendLambda
10. //不需要线程调度-返回SuspendLambda
11. intercepted?:(context[ContinuationInterceptor]?.interceptContinuation(this)?:this).also{intercepted=it}
13. //CoroutineDispatcher
14. //continuation-SuspendLambda->ContinuationImpl->BaseContinuationImpl
15. publicfinaloverridefun<T>interceptContinuation(continuation:Continuation<T>):Continuation<T>=
16. DispatchedContinuation(this,continuation)

接下来看看 resumeCancellableWith 是怎么启动协程的，这里还涉及到Dispatchers线程调度的逻辑：

1. internalclassDispatchedContinuation<inT>(
3. valdispatcher:CoroutineDispatcher,
5. valcontinuation:Continuation<T>
6. ):DispatchedTask<T>(MODE_ATOMIC_DEFAULT),CoroutineStackFrame,Continuation<T>bycontinuation{
7. publicfun<T>Continuation<T>.resumeCancellableWith(result:Result<T>):Unit=when(this){
8. //进行线程调度，最后也会执行到continuation.resumeWith方法
9. isDispatchedContinuation->resumeCancellableWith(result)
10. //直接执行continuation.resumeWith方法
11. else->resumeWith(result)
12. }
14. inlinefunresumeCancellableWith(result:Result<T>){
15. valstate=result.toState()
16. //判断是否需要线程调度
17. if(dispatcher.isDispatchNeeded(context)){
18. _state=state
19. resumeMode=MODE_CANCELLABLE
20. //需要调度则先进行调度
21. dispatcher.dispatch(context,this)
22. }else{
23. executeUnconfined(state,MODE_CANCELLABLE){
24. if(!resumeCancelled()){
25. //不需要调度则直接在当前线程执行协程
26. resumeUndispatchedWith(result)
27. }
28. }
29. }
30. }
32. inlinefunresumeUndispatchedWith(result:Result<T>){
33. withCoroutineContext(context,countOrElement){
34. continuation.resumeWith(result)
35. }
36. }
37. }

• 当需要线程调度时，则在调度后会调用 DispatchedContinuation.continuation.resumeWith 来启动协程，其中 continuation 是 SuspendLambda 实例;
• 当不需要线程调度时，则直接调用 SuspendLambda.resumeWith 来启动协程。

resumeWith 方法调用的是父类 BaseContinuationImpl 中的 resumeWith 方法：

1. internalabstractclassBaseContinuationImpl(publicvalcompletion:Continuation<Any?>?):Continuation<Any?>,CoroutineStackFrame,Serializable{
2. publicfinaloverridefunresumeWith(result:Result<Any?>){
3. //…
4. valoutcome=invokeSuspend(param)
5. //…
6. }
7. }

因此，协程的启动是通过 BaseContinuationImpl.resumeWith 方法调用到了子类 SuspendLambda.invokeSuspend 方法，然后通过状态机来控制顺序运行。

协程的挂起和恢复

Kotlin 编译器会为 协程体 生成继承自 SuspendLambda 的子类，协程的真正运算逻辑都在其 invokeSuspend 方法中。上一节介绍了 launch 是怎么创建和启动协程的，在这一节我们再看看当协程代码执行到 suspend 函数后，协程是怎么被挂起的 以及 当 suspend 函数执行完成得到可用结果后是怎么恢复协程的。

Kotlin 协程的内部实现使用了 Kotlin 编译器的一些编译技术，当 suspend 函数被调用时，都有一个隐式的参数额外传入，这个参数是 Continuation 类型，封装了协程 resume 后执行的代码逻辑。

1. privatesuspendfungetId():String{
2. returnGlobalScope.async(Dispatchers.IO){
3. delay(1000)
4. "hearing"
5. }.await()
6. }
8. //Decompile成Java
9. finalObjectgetId(
11. Continuation$completion){
12. //…
13. }

其中传入的 $completion 参数，从上一节可以看到是调用 getId 方法所在的协程体对象，也就是一个 SuspendLambda 对象。Continuation的定义如下：

1. publicinterfaceContinuation<inT>{
2. publicvalcontext:CoroutineContext
4. publicfunresumeWith(result:Result<T>)
5. }

将 getId 方法编译后的字节码反编译成 Java 代码如下(为便于阅读，删减及修改了部分代码)：

1. finalObjectgetId(
3. Continuation$completion){
4. //新建与启动协程
5. returnBuildersKt.async$default((CoroutineScope)GlobalScope.INSTANCE,(CoroutineContext)Dispatchers.getIO(),(CoroutineStart)null,(Function2)(newFunction2((Continuation)null){
6. intlabel;
8. publicfinalObjectinvokeSuspend(
10. Object$result){
11. switch(this.label){
12. case0:
13. ResultKt.throwOnFailure($result);
14. this.label=1;
15. if(DelayKt.delay(1000L,this)==COROUTINE_SUSPENDED){
16. returnCOROUTINE_SUSPENDED;
17. }
18. break;
19. case1:
20. ResultKt.throwOnFailure($result);
21. break;
22. default:
23. thrownewIllegalStateException("callto'resume'before'invoke'withcoroutine");
24. }
25. return"hearing";
26. }
28. //…
29. }),2,(Object)null).await($completion);//调用await()suspend函数
30. }

结合协程的状态机一节，当上面的 launch 协程体执行到 getId 方法时, 会根据其返回值是否为 COROUTINE_SUSPENDED 来决定是否挂起，由于 getId 的逻辑是通过 async 启动一个新的协程，协程体内调用了 suspend delay 方法，然后通过 await suspend 函数等待结果，当 async 协程没完成时, await 会返回 COROUTINE_SUSPENDED, 因此 launch 协程体的 invokeSuspend 方法直接 return COROUTINE_SUSPENDED 值执行完成，此时 launch 启动的协程处于挂起状态但不阻塞所处线程，而 async 启动的协程开始执行。

我们看一下 async 的源码：

1. publicfun<T>CoroutineScope.async(…):Deferred<T>{
2. valnewContext=newCoroutineContext(context)
3. valcoroutine=if(start.isLazy)LazyDeferredCoroutine(newContext,block)else
4. DeferredCoroutine<T>(newContext,active=true)
5. coroutine.start(start,coroutine,block)
6. returncoroutine
7. }

默认情况下，上面的 coroutine 取 DeferredCoroutine 实例，于是我们看一下其 await 方法以及在 async 协程执行完成后，是怎么恢复 launch 协程的：

1. privateopenclassDeferredCoroutine<T>(
2. parentContext:CoroutineContext,active:Boolean
3. ):AbstractCoroutine<T>(parentContext,active),Deferred<T>,SelectClause1<T>{
4. overridesuspendfunawait():T=awaitInternal()asT
5. }
7. //JobSupport
8. internalsuspendfunawaitInternal():Any?{
9. while(true){//lock-freelooponstate
10. valstate=this.state
11. if(state!isIncomplete){
12. //已经完成，则直接返回结果
13. if(stateisCompletedExceptionally){//Slowpathtorecoverstacktrace
14. recoverAndThrow(state.cause)
15. }
16. returnstate.unboxState()
17. }
18. //不需要重试时直接break，执行awaitSuspend
19. if(startInternal(state)>=0)break
20. }
21. returnawaitSuspend()//slow-path
22. }
24. //suspendCoroutineUninterceptedOrReturn:获取当前协程，且挂起当前协程(返回COROUTINE_SUSPENDED)或不挂起直接返回结果
25. privatesuspendfunawaitSuspend():Any?=suspendCoroutineUninterceptedOrReturn{uCont->
26. valcont=AwaitContinuation(uCont.intercepted(),this)
27. cont.disposeOnCancellation(invokeOnCompletion(ResumeAwaitOnCompletion(this,cont).asHandler))
28. cont.getResult()
29. }

上面 awaitInternal 的大致逻辑是当挂起函数已经有结果时则直接返回，否则挂起父协程，然后 invokeOnCompletion 方法将 ResumeAwaitOnCompletion 插入一个队列(state.list)中，源码就不再贴出了。接着看看在 async 执行完成后是怎么调用 ResumeAwaitOnCompletion 来 resume 被挂起的协程的。注意：不要绕进 async 协程体中 delay 是怎么挂起和恢复 async 协程的这一逻辑，我们不需要关注这一层!

接着 async 协程的执行往下看，从前面可知它会调用 BaseContinuationImpl.resumeWith 方法来执行协程逻辑，我们详细看一下这个方法，在这里会执行该协程的 invokeSuspend 函数：

1. internalabstractclassBaseContinuationImpl(
2. publicvalcompletion:Continuation<Any?>?
3. ):Continuation<Any?>,CoroutineStackFrame,Serializable{
4. publicfinaloverridefunresumeWith(result:Result<Any?>){
5. varcurrent=this
6. varparam=result
7. while(true){
8. with(current){
9. valcompletion=completion!!//failfastwhentryingtoresumecontinuationwithoutcompletion
10. valoutcome:Result<Any?>=
11. try{//调用invokeSuspend方法执行协程逻辑
12. valoutcome=invokeSuspend(param)
13. //协程挂起时返回的是COROUTINE_SUSPENDED，即协程挂起时，resumeWith执行结束
14. //再次调用resumeWith时协程挂起点之后的代码才能继续执行
15. if(outcome===COROUTINE_SUSPENDED)return
16. Result.success(outcome)
17. }catch(exception:Throwable){
18. Result.failure(exception)
19. }
20. releaseIntercepted()//thisstatemachineinstanceisterminating
21. if(completionisBaseContinuationImpl){
22. //unrollingrecursionvialoop
23. current=completion
24. param=outcome
25. }else{
26. //top–levelcompletionreached–invokeandreturn
27. completion.resumeWith(outcome)
28. return
29. }
30. }
31. }
32. }
33. }

我们从上面的源码可以看到，在 createCoroutineUnintercepted 方法中创建的 SuspendLambda 实例是 BaseContinuationImpl 的子类对象，其 completion 参数为下：

• launch: if (isLazy) LazyStandaloneCoroutine else StandaloneCoroutine
• async: if (isLazy) LazyDeferredCoroutine else DeferredCoroutine

上面这几个类都是 AbstractCoroutine 的子类。而根据 completion 的类型会执行不同的逻辑：

• BaseContinuationImpl: 执行协程逻辑
• 其它: 调用 resumeWith 方法，处理协程的状态，协程挂起后的恢复即与它有关

在上面的例子中 async 启动的协程，它也会调用其 invokeSuspend 方法执行 async 协程逻辑，假设 async 返回的结果已经可用时，即非 COROUTINE_SUSPENDED 值，此时 completion 是 DeferredCoroutine 对象，因此会调用 DeferredCoroutine.resumeWith 方法，然后返回，父协程的恢复逻辑便是在这里。

1. //AbstractCoroutine
2. publicfinaloverridefunresumeWith(result:Result<T>){
3. valstate=makeCompletingOnce(result.toState())
4. if(state===COMPLETING_WAITING_CHILDREN)return
5. afterResume(state)
6. }

在 makeCompletingOnce 方法中，会根据 state 去处理协程状态，并执行上面插入 state.list 队列中的 ResumeAwaitOnCompletion.invoke 来恢复父协程，必要的话还会把 async 的结果给它，具体代码实现太多就不贴了，不是本节的重点。直接看 ResumeAwaitOnCompletion.invoke 方法：

1. privateclassResumeAwaitOnCompletion<T>(
2. job:JobSupport,privatevalcontinuation:CancellableContinuationImpl<T>
3. ):JobNode<JobSupport>(job){
4. overridefuninvoke(cause:Throwable?){
5. valstate=job.state
6. assert{state!isIncomplete}
7. if(stateisCompletedExceptionally){
8. //Resumewithwiththecorrespondingexceptiontopreserveit
9. continuation.resumeWithException(state.cause)
10. }else{
11. //resume被挂起的协程
12. continuation.resume(state.unboxState()asT)
13. }
14. }
15. }

这里的 continuation 就是 launch 协程体，也就是 SuspendLambda 对象，于是 invoke 方法会再一次调用到 BaseContinuationImpl.resumeWith 方法，接着调用 SuspendLambda.invokeSuspend, 然后根据 label 取值继续执行接下来的逻辑!

suspendCoroutineUninterceptedOrReturn

接下来我们看一下怎么将一个基于回调的方法改造成一个基于协程的 suspend 方法，要实现这个需求，重点在于 suspendCoroutineUninterceptedOrReturn 方法，根据注释，这个方法的作用是: Obtains the current continuation instance inside suspend functions and either suspends currently running coroutine or returns result immediately without suspension. 即获取当前协程的实例，并且挂起当前协程或不挂起直接返回结果。函数定义如下：

1. publicsuspendinlinefun<T>suspendCoroutineUninterceptedOrReturn(crossinlineblock:(Continuation<T>)->Any?):T{
2. //…
3. }

根据 block 的返回值，有两种情况：

• 如果 block 返回 COROUTINE_SUSPENDED, 意味着 suspend 函数会挂起当前协程而不会立即返回结果。这种情况下, block 中的 Continuation 需要在结果可用后调用 Continuation.resumeWith 来 resume 协程。
• 如果 block 返回的 T 是 suspend 函数的结果，则协程不会被挂起, block 中的 Continuation 不会被调用。

调用 Continuation.resumeWith 会直接在调用者的线程 resume 协程，而不会经过 CoroutineContext 中可能存在的 ContinuationInterceptor。建议使用更安全的 suspendCoroutine 方法，在其 block 中可以同步或在异步线程调用 Continuation.resume 和 Continuation.resumeWithException:

1. publicsuspendinlinefun<T>suspendCoroutine(crossinlineblock:(Continuation<T>)->Unit):T{
2. contract{callsInPlace(block,InvocationKind.EXACTLY_ONCE)}
3. returnsuspendCoroutineUninterceptedOrReturn{c:Continuation<T>->
4. //调用拦截器
5. valsafe=SafeContinuation(c.intercepted())
6. block(safe)
7. safe.getOrThrow()
8. }
9. }

此外除了 suspendCoroutine 方法，还有 suspendCancellableCoroutine, suspendAtomicCancellableCoroutine, suspendAtomicCancellableCoroutineReusable 等方法都可以用来将异步回调的方法封装成 suspend 函数。

下面来看一个例子来介绍怎么将异步回调函数封装成 suspend 函数：

1. classNetFetcher{
2. //将下面的request方法封装成suspend方法
3. suspendfunrequestSuspend(id:Int):String=suspendCoroutine{continuation->
4. request(id,object:OnResponseListener{
5. overridefunonResponse(response:String){
6. continuation.resume(response)
7. }
9. overridefunonError(error:String){
10. continuation.resumeWithException(Exception(error))
11. }
12. })
13. }
15. funrequest(id:Int,listener:OnResponseListener){
16. Thread.sleep(5000)
17. if(id%2==0){
18. listener.onResponse("success")
19. }else{
20. listener.onError("error")
21. }
22. }
24. interfaceOnResponseListener{
25. funonResponse(response:String)
26. funonError(error:String)
27. }
28. }
30. objectMain{
31. funmain(){
32. requestByCoroutine()
33. }
35. //使用回调
36. privatefunrequestByCallback(){
37. NetFetcher().request(21,object:NetFetcher.OnResponseListener{
38. overridefunonResponse(response:String){
39. println("result=$response")
40. }
42. overridefunonError(error:String){
43. println("result=$error")
44. }
45. })
46. }
48. //使用协程
49. privatefunrequestByCoroutine(){
50. GlobalScope.launch(Dispatchers.Main){
51. valresult=withContext(Dispatchers.IO){
52. try{
53. NetFetcher().requestSuspend(22)
54. }catch(e:Exception){
55. e.message
56. }
57. }

为加深理解，再介绍一下 Kotlin 提供的两个借助 suspendCancellableCoroutine 实现的挂起函数: delay & yield。

delay

delay 方法借助了 suspendCancellableCoroutine 方法来挂起协程：

1. publicsuspendfundelay(timeMillis:Long){
2. if(timeMillis<=0)return//don'tdelay
3. returnsuspendCancellableCoroutinesc@{cont:CancellableContinuation<Unit>->
4. cont.context.delay.scheduleResumeAfterDelay(timeMillis,cont)
5. }
6. }
8. overridefunscheduleResumeAfterDelay(timeMillis:Long,continuation:CancellableContinuation<Unit>){
9. postDelayed(Runnable{
10. with(continuation){resumeUndispatched(Unit)}
11. },timeMillis)
12. }

可以看出这里 delay 的逻辑类似于 Handle 机制，将 resumeUndispatched 封装的 Runnable 放到一个队列中，在延迟的时间到达便会执行 resume 恢复协程。

yield

yield 方法作用是挂起当前协程，这样可以让该协程所在线程运行其他逻辑，当其他协程执行完成或也调用 yield 让出执行权时，之前的协程可以恢复执行。

1. launch(Dispatchers.Main){
2. repeat(3){
3. println("job1$it")
4. yield()
5. }
6. }
7. launch(Dispatchers.Main){
8. repeat(3){
9. println("job2$it")
10. yield()
11. }
12. }
14. //output
15. job10
16. job20
17. job11
18. job21
19. job12
20. job22

看一下 yield 的源码：

1. publicsuspendfunyield():Unit=suspendCoroutineUninterceptedOrReturnsc@{uCont->
2. valcontext=uCont.context
3. //如果协程没有调度器，或者像Unconfined一样没有进行调度则直接返回
4. valcont=uCont.intercepted()as?DispatchedContinuation<Unit>?:return@scUnit
5. if(cont.dispatcher.isDispatchNeeded(context)){
6. //thisisaregulardispatcher–dosimpledispatchYield
7. cont.dispatchYield(context,Unit)
8. }else{
9. //Thisiseitheran"immediate"dispatcherortheUnconfineddispatcher
10. //…
11. }
12. COROUTINE_SUSPENDED
13. }
15. //DispatchedContinuation
16. internalfundispatchYield(context:CoroutineContext,value:T){
17. _state=value
18. resumeMode=MODE_CANCELLABLE
19. dispatcher.dispatchYield(context,this)
20. }
22. publicopenfundispatchYield(context:CoroutineContext,block:Runnable):Unit=dispatch(context,block)

可知 dispatchYield 会调用到 dispatcher.dispatch 方法将协程分发到调度器队列中，这样线程可以执行其他协程，等到调度器再次执行到该协程时，会 resume 该协程。

总结

通过上面协程的工作原理解析，可以从源码中发现 Kotlin 中的协程存在着三层包装：

• 第一层包装: launch & async 返回的 Job, Deferred 继承自 AbstractCoroutine, 里面封装了协程的状态，提供了 cancel 等接口;
• 第二层包装: 编译器生成的 SuspendLambda 子类，封装了协程的真正执行逻辑，其继承关系为 SuspendLambda -> ContinuationImpl -> BaseContinuationImpl, 它的 completion 参数就是第一层包装实例;
• 第三层包装: DispatchedContinuation, 封装了线程调度逻辑，它的 continuation 参数就是第二层包装实例。

这三层包装都实现了 Continuation 接口，通过代理模式将协程的各层包装组合在一起，每层负责不同的功能，如下图：

原文地址：https://ljd1996.github.io/2021/05/19/Kotlin%E7%AC%94%E8%AE%B0%E4%B9%8B%E5%8D%8F%E7%A8%8B%E5%B7%A5%E4%BD%9C%E5%8E%9F%E7%90%86/?utm_source=tuicool&utm_medium=referral