破解 Kotlin 协程(6) - 协程挂起篇

╰+哭是因爲堅強的太久メ 2022-04-25 07:08 223阅读 0赞

**关键词：Kotlin 协程 协程挂起 任务挂起 suspend 非阻塞**

> 协程的挂起最初是一个很神秘的东西，因为我们总是用线程的概念去思考，所以我们只能想到阻塞。不阻塞的挂起到底是怎么回事呢？说出来你也许会笑~~（哭？。。抱歉这篇文章我实在是没办法写的更通俗易懂了，大家一定要亲手实践！）

## 1. 先看看 delay ##

我们刚刚学线程的时候，最常见的模拟各种延时用的就是 `Thread.sleep` 了，而在协程里面，对应的就是 `delay`。 `sleep` 让线程进入休眠状态，直到指定时间之后某种信号或者条件到达，线程就尝试恢复执行，而 `delay` 会让协程挂起，这个过程并不会阻塞 CPU，甚至可以说从硬件使用效率上来讲是“什么都不耽误”，从这个意义上讲 `delay` 也可以是让协程休眠的一种很好的手段。

`delay` 的源码其实很简单：

1.  `public suspend fun delay(timeMillis: Long) {`
2.  `if (timeMillis <= 0) return // don't delay`
3.  `return suspendCancellableCoroutine sc@ { cont: CancellableContinuation<Unit> ->`
4.  `cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)`
5.  `}`
6.  `}`

`cont.context.delay.scheduleResumeAfterDelay` 这个操作，你可以类比 JavaScript 的 `setTimeout`，Android 的 `handler.postDelay`，本质上就是设置了一个延时回调，时间一到就调用 `cont` 的 resume 系列方法让协程继续执行。

剩下的最关键的就是 `suspendCancellableCoroutine` 了，这可是我们的老朋友了，前面我们用它实现了回调到协程的各种转换 —— 原来 `delay` 也是基于它实现的，如果我们再多看一些源码，你就会发现类似的还有 `join`、 `await` 等等。

## 2. 再来说说 suspendCancellableCoroutine ##

既然大家对于 `suspendCancellableCoroutine` 已经很熟悉了，那么我们干脆直接召唤一个老朋友给大家：

1.  `private suspend fun joinSuspend() = suspendCancellableCoroutine<Unit> { cont ->`
2.  `cont.disposeOnCancellation(invokeOnCompletion(handler = ResumeOnCompletion(this, cont).asHandler))`
3.  `}`

`Job.join()` 这个方法会首先检查调用者 `Job` 的状态是否已经完成，如果是，就直接返回并继续执行后面的代码而不再挂起，否则就会走到这个 `joinSuspend` 的分支当中。我们看到这里只是注册了一个完成时的回调，那么传说中的 `suspendCancellableCoroutine` 内部究竟做了什么呢？

1.  `public suspend inline fun <T> suspendCancellableCoroutine(`
2.  `crossinline block: (CancellableContinuation<T>) -> Unit`
3.  `): T =`
4.  `suspendCoroutineUninterceptedOrReturn { uCont ->`
5.  `val cancellable = CancellableContinuationImpl(uCont.intercepted(), resumeMode = MODE_CANCELLABLE)`
6.  `block(cancellable)`
7.  `cancellable.getResult() // 这里的类型是 Any?`
8.  `}`

`suspendCoroutineUninterceptedOrReturn` 这个方法调用的源码是看不到的，因为它根本没有源码：P 它的逻辑就是帮大家拿到 `Continuation` 实例，真的就只有这样。不过这样说起来还是很抽象，因为有一处非常的可疑： `suspendCoroutineUninterceptedOrReturn` 的返回值类型是 `T`，而传入的 lambda 的返回值类型是 `Any?`， 也就是我们看到的 `cancellable.getResult()` 的类型是 `Any?`，这是为什么？

我记得在协程系列文章的开篇，我就提到过 `suspend` 函数的签名，当时是以 `await` 为例的，这个方法大致相当于：

1.  `fun await(continuation: Continuation<User>): Any {`
2.  `...`
3.  `}`

`suspend` 一方面为这个方法添加了一个 `Continuation` 的参数，另一方面，原先的返回值类型 `User` 成了 `Continuation` 的泛型实参，而真正的返回值类型竟然是 `Any`。当然，这里因为定义的逻辑返回值类型 `User`是不可空的，因此真实的返回值类型也用了 `Any` 来示意，如果泛型实参是个可空的类型，那么真实的返回值类型也就是 `Any?` 了，这正与前面提到的 `cancellable.getResult()` 返回的这个 `Any?` 相对应。

> 如果大家去查 `await` 的源码，你同样会看到这个 `getResult()` 的调用。

简单来说就是，对于 `suspend` 函数，不是一定要挂起的，可以在需要的时候挂起，也就是要等待的协程还没有执行完的时候，等待协程执行完再继续执行；而如果在开始 `join` 或者 `await` 或者其他 `suspend` 函数，如果目标协程已经完成，那么就没必要等了，直接拿着结果走人即可。那么这个神奇的逻辑就在于 `cancellable.getResult()` 究竟返回什么了，且看：

1.  `internal fun getResult(): Any? {`
2.  `...`
3.  `if (trySuspend()) return COROUTINE_SUSPENDED // ① 触发挂起逻辑`
4.  `...`
5.  `if (state is CompletedExceptionally) // ② 异常立即抛出`
6.  `throw recoverStackTrace(state.cause, this)`
7.  `return getSuccessfulResult(state) // ③ 正常结果立即返回`
8.  `}`

这段代码 ① 处就是挂起逻辑了，表示这时候目标协程还没有执行完，需要等待结果，②③是协程已经执行完可以直接拿到异常和正常结果的两种情况。②③好理解，关键是 ①，它要挂起，这返回的是个什么东西？

1.  `public val COROUTINE_SUSPENDED: Any get() = CoroutineSingletons.COROUTINE_SUSPENDED`
2.   
3.  `internal enum class CoroutineSingletons { COROUTINE_SUSPENDED, UNDECIDED, RESUMED }`

这是 1.3 的实现，1.3 以前的实现更有趣，就是一个白板 `Any`。其实是什么不重要，关键是这个东西是一个单例，任何时候协程见到它就知道自己该挂起了。

## 3. 深入挂起操作 ##

既然说到挂起，大家可能觉得还是一知半解，还是不知道挂起究竟怎么做到的，怎么办？说真的这个挂起是个什么操作其实一直没有拿出来给大家看，不是我们太小气了，只是太早拿出来会比较吓人。。

1.  `suspend fun hello() = suspendCoroutineUninterceptedOrReturn<Int>{`
2.  `continuation ->`
3.  `log(1)`
4.  `thread {`
5.  `Thread.sleep(1000)`
6.  `log(2)`
7.  `continuation.resume(1024)`
8.  `}`
9.  `log(3)`
10. `COROUTINE_SUSPENDED`
11. `}`

我写了这么一个 `suspend` 函数，在 `suspendCoroutineUninterceptedOrReturn` 当中直接返回了这个传说中的白板 `COROUTINE_SUSPENDED`，正常来说我们应该在一个协程当中调用这个方法对吧，可是我偏不，我写一段 Java 代码去调用这个方法，结果会怎样呢？

1.  `public class CallCoroutine {`
2.  `public static void main(String... args) {`
3.  `Object value = SuspendTestKt.hello(new Continuation<Integer>() {`
4.  `@NotNull`
5.  `@Override`
6.  `public CoroutineContext getContext() {`
7.  `return EmptyCoroutineContext.INSTANCE;`
8.  `}`
9.   
10. `@Override`
11. `public void resumeWith(@NotNull Object o) { // ①`
12. `if(o instanceof Integer){`
13. `handleResult(o);`
14. `} else {`
15. `Throwable throwable = (Throwable) o;`
16. `throwable.printStackTrace();`
17. `}`
18. `}`
19. `});`
20.  
21. `if(value == IntrinsicsKt.getCOROUTINE_SUSPENDED()){ // ②`
22. `LogKt.log("Suspended.");`
23. `} else {`
24. `handleResult(value);`
25. `}`
26. `}`
27.  
28. `public static void handleResult(Object o){`
29. `LogKt.log("The result is " + o);`
30. `}`
31. `}`

这段代码看上去比较奇怪，可能会让人困惑的有两处：

① 处，我们在 Kotlin 当中看到的 `resumeWith` 的参数类型是 `Result`，怎么这儿成了 `Object` 了？因为 `Result` 是内联类，编译时会用它唯一的成员替换掉它，因此就替换成了 `Object` （在Kotlin 里面是 `Any?`）

② 处 `IntrinsicsKt.getCOROUTINE_SUSPENDED()` 就是 Kotlin 的 `COROUTINE_SUSPENDED`

剩下的其实并不难理解，运行结果自然就是如下所示了：

1.  `07:52:55:288 [main] 1`
2.  `07:52:55:293 [main] 3`
3.  `07:52:55:296 [main] Suspended.`
4.  `07:52:56:298 [Thread-0] 2`
5.  `07:52:56:306 [Thread-0] The result is 1024`

其实这段 Java 代码的调用方式与 Kotlin 下面的调用已经很接近了：

1.  `suspend fun main() {`
2.  `log(hello())`
3.  `}`

只不过我们在 Kotlin 当中还是不太容易拿到 `hello` 在挂起时的真正返回值，其他的返回结果完全相同。

1.  `12:44:08:290 [main] 1`
2.  `12:44:08:292 [main] 3`
3.  `12:44:09:296 [Thread-0] 2`
4.  `12:44:09:296 [Thread-0] 1024`

很有可能你看到这里都会觉得晕头转向，没有关系，我现在已经开始尝试揭示一些协程挂起的背后逻辑了，比起简单的使用，概念的理解和接受需要有个小小的过程。

## 4. 深入理解协程的状态转移 ##

前面我们已经对协程的原理做了一些揭示，显然 Java 的代码让大家能够更容易理解，那么接下来我们再来看一个更复杂的例子：

1.  `suspend fun returnSuspended() = suspendCoroutineUninterceptedOrReturn<String>{`
2.  `continuation ->`
3.  `thread {`
4.  `Thread.sleep(1000)`
5.  `continuation.resume("Return suspended.")`
6.  `}`
7.  `COROUTINE_SUSPENDED`
8.  `}`
9.   
10. `suspend fun returnImmediately() = suspendCoroutineUninterceptedOrReturn<String>{`
11. `log(1)`
12. `"Return immediately."`
13. `}`

我们首先定义两个挂起函数，第一个会真正挂起，第二个则会直接返回结果，这类似于我们前面讨论 `join` 或者 `await` 的两条路径。我们再用 Kotlin 给出一个调用它们的例子：

1.  `suspend fun main() {`
2.  `log(1)`
3.  `log(returnSuspended())`
4.  `log(2)`
5.  `delay(1000)`
6.  `log(3)`
7.  `log(returnImmediately())`
8.  `log(4)`
9.  `}`

运行结果如下：

1.  `08:09:37:090 [main] 1`
2.  `08:09:38:096 [Thread-0] Return suspended.`
3.  `08:09:38:096 [Thread-0] 2`
4.  `08:09:39:141 [kotlinx.coroutines.DefaultExecutor] 3`
5.  `08:09:39:141 [kotlinx.coroutines.DefaultExecutor] Return immediately.`
6.  `08:09:39:141 [kotlinx.coroutines.DefaultExecutor] 4`

好，现在我们要揭示这段协程代码的真实面貌，为了做到这一点，我们用 Java 来仿写一下这段逻辑：

> 注意，下面的代码逻辑上并不能做到十分严谨，不应该出现在生产当中，仅供学习理解协程使用。

1.  `public class ContinuationImpl implements Continuation<Object> {`
2.   
3.  `private int label = 0;`
4.  `private final Continuation<Unit> completion;`
5.   
6.  `public ContinuationImpl(Continuation<Unit> completion) {`
7.  `this.completion = completion;`
8.  `}`
9.   
10. `@Override`
11. `public CoroutineContext getContext() {`
12. `return EmptyCoroutineContext.INSTANCE;`
13. `}`
14.  
15. `@Override`
16. `public void resumeWith(@NotNull Object o) {`
17. `try {`
18. `Object result = o;`
19. `switch (label) {`
20. `case 0: {`
21. `LogKt.log(1);`
22. `result = SuspendFunctionsKt.returnSuspended( this);`
23. `label++;`
24. `if (isSuspended(result)) return;`
25. `}`
26. `case 1: {`
27. `LogKt.log(result);`
28. `LogKt.log(2);`
29. `result = DelayKt.delay(1000, this);`
30. `label++;`
31. `if (isSuspended(result)) return;`
32. `}`
33. `case 2: {`
34. `LogKt.log(3);`
35. `result = SuspendFunctionsKt.returnImmediately( this);`
36. `label++;`
37. `if (isSuspended(result)) return;`
38. `}`
39. `case 3:{`
40. `LogKt.log(result);`
41. `LogKt.log(4);`
42. `}`
43. `}`
44. `completion.resumeWith(Unit.INSTANCE);`
45. `} catch (Exception e) {`
46. `completion.resumeWith(e);`
47. `}`
48. `}`
49.  
50. `private boolean isSuspended(Object result) {`
51. `return result == IntrinsicsKt.getCOROUTINE_SUSPENDED();`
52. `}`
53. `}`

我们定义了一个 Java 类 `ContinuationImpl`，它就是一个 `Continuation` 的实现。

> 实际上如果你愿意，你还真得可以在 Kotlin 的标准库当中找到一个名叫 `ContinuationImpl` 的类，只不过，它的 `resumeWith` 最终调用到了 `invokeSuspend`，而这个 `invokeSuspend` 实际上就是我们的协程体，通常也就是一个 Lambda 表达式 —— 我们通过 `launch`启动协程，传入的那个 Lambda 表达式，实际上会被编译成一个 `SuspendLambda` 的子类，而它又是 `ContinuationImpl` 的子类。

有了这个类我们还需要准备一个 completion 用来接收结果，这个类仿照标准库的 `RunSuspend` 类实现，如果你有阅读前面的文章，那么你应该知道 suspend main 的实现就是基于这个类：

1.  `public class RunSuspend implements Continuation<Unit> {`
2.   
3.  `private Object result;`
4.   
5.  `@Override`
6.  `public CoroutineContext getContext() {`
7.  `return EmptyCoroutineContext.INSTANCE;`
8.  `}`
9.   
10. `@Override`
11. `public void resumeWith(@NotNull Object result) {`
12. `synchronized (this){`
13. `this.result = result;`
14. `notifyAll(); // 协程已经结束，通知下面的 wait() 方法停止阻塞`
15. `}`
16. `}`
17.  
18. `public void await() throws Throwable {`
19. `synchronized (this){`
20. `while (true){`
21. `Object result = this.result;`
22. `if(result == null) wait(); // 调用了 Object.wait()，阻塞当前线程，在 notify 或者 notifyAll 调用时返回`
23. `else if(result instanceof Throwable){`
24. `throw (Throwable) result;`
25. `} else return;`
26. `}`
27. `}`
28. `}`
29. `}`

这段代码的关键点在于 `await()` 方法，它在其中起了一个死循环，不过大家不要害怕，这个死循环是个纸老虎，如果 `result` 是 `null`，那么当前线程会被立即阻塞，直到结果出现。具体的使用方法如下：

1.  `...`
2.  `public static void main(String... args) throws Throwable {`
3.  `RunSuspend runSuspend = new RunSuspend();`
4.  `ContinuationImpl table = new ContinuationImpl(runSuspend);`
5.  `table.resumeWith(Unit.INSTANCE);`
6.  `runSuspend.await();`
7.  `}`
8.  `...`

> 这写法简直就是 suspend main 的真实面貌了。

我们看到，作为 completion 传入的 `RunSuspend` 实例的 `resumeWith` 实际上是在 `ContinuationImpl` 的 `resumeWtih` 的最后才会被调用，因此它的 `await()` 一旦进入阻塞态，直到 `ContinuationImpl` 的整体状态流转完毕才会停止阻塞，此时进程也就运行完毕正常退出了。

于是这段代码的运行结果如下：

1.  `08:36:51:305 [main] 1`
2.  `08:36:52:315 [Thread-0] Return suspended.`
3.  `08:36:52:315 [Thread-0] 2`
4.  `08:36:53:362 [kotlinx.coroutines.DefaultExecutor] 3`
5.  `08:36:53:362 [kotlinx.coroutines.DefaultExecutor] Return immediately.`
6.  `08:36:53:362 [kotlinx.coroutines.DefaultExecutor] 4`

我们看到，这段普通的 Java 代码与前面的 Kotlin 协程调用完全一样。那么我这段 Java 代码的编写根据是什么呢？就是 Kotlin 协程编译之后产生的字节码。当然，字节码是比较抽象的，我这样写出来就是为了让大家更容易的理解协程是如何执行的，看到这里，相信大家对于协程的本质有了进一步的认识：

*  协程的挂起函数本质上就是一个回调，回调类型就是 `Continuation`
 *  协程体的执行就是一个状态机，每一次遇到挂起函数，都是一次状态转移，就像我们前面例子中的 `label` 不断的自增来实现状态流转一样

如果能够把这两点认识清楚，那么相信你在学习协程其他概念的时候就都将不再是问题了。如果想要进行线程调度，就按照我们讲到的调度器的做法，在 `resumeWith` 处执行线程切换就好了，其实非常容易理解的。官方的协程框架本质上就是在做这么几件事儿，如果你去看源码，可能一时云里雾里，主要是因为框架除了实现核心逻辑外还需要考虑跨平台实现，还需要优化性能，但不管怎么样，这源码横竖看起来就是五个字：状态机回调。

## 5. 小结 ##

不同以往，我们从这一篇开始毫无保留的为大家尝试揭示协程背后的逻辑，也许一时间可能有些难懂，不过不要紧，你可以使用协程一段时间之后再来阅读这些内容，相信一定会豁然开朗的。

当然，这一篇内容的安排更多是为后面的序列篇开路，Kotlin 的 `Sequence` 就是基于协程实现的，它的用法很简单，几乎与普通的 `Iterable` 没什么区别，因此序列篇我们会重点关注它的内部实现原理，欢迎大家关注。

另外，想要找到好 Offer、想要实现技术进阶的迷茫中的 Android 工程师们，推荐大家关注下我的新课《破解Android高级面试》，这门课已经更新完毕，涉及内容均非浅尝辄止，目前已经有300+同学在学习，你还在等什么(\*≧∪≦)：

**长按识别二维码即可进入课程啦！**

**![640?wx\_fmt=jpeg][640_wx_fmt_jpeg]**

[640_wx_fmt_jpeg]: /images/20220125/0444951993344b249a59d9b9688aad9b.png