Kotlin Coroutine suspend 原理解析

一. 回调地狱

1.1 同步操作

假如我们有这样一个需求:

fun childFun1(): Int {
    return 10
}

fun childFun2(): Int {
    return 20
}

fun childFun3(num1: Int, num2: Int): Int {
    return num1 + num2
}

fun parentFun() : Int {
    val num1 = childFun1()
    val num2 = childFun2()
    val sum = childFun3(num1, num2)
    return sum
}

即: 从多个操作中(childFun1, childFun2) 获取值，然后再对这些值进行处理(childFun3)，程序逻辑非常直观易懂。

例如，先上传用户选择的图片到服务端储存，获取服务端返回的图片对应的地址，然后这些地址设置到对应位置。

1.2 异步操作

如果这些操作都是耗时操作，为了不阻塞线程，需要将这些耗时操作放到其他线程中，即

val executor: ExecutorService = Executors.newFixedThreadPool(2)

fun childFun1(callback: (Int) -> Unit): Unit {
    executor.execute {
        Thread.sleep(1000)
        callback(10)
    }
}

fun childFun2(callback: (Int) -> Unit): Unit {
    executor.execute {
        Thread.sleep(2000)
        callback(20)
    }
}

fun childFun3(num1: Int, num2: Int, callback: (Int) -> Unit): Unit {
    executor.execute {
        Thread.sleep(500)
        callback(num1 + num2)
    }
}

fun parentFun(callback: (Int) -> Unit) : Unit {
    childFun1(fun(num1) {
        childFun2(fun(num2) {
            childFun3(num1, num2, callback)
        })
    })
}

因为是异步操作，结果值不能直接返回，只能通过 callback 方式异步回传，所以当异步操作很多的时候，整个回调链就很长了，让代码逻辑显得不清晰。

1.3 协程

suspend fun childFun1(): Int {
    delay(1000)
    return 10
}

suspend fun childFun2(): Int {
    delay(2000)
    return 20
}

suspend fun childFun3(num1: Int, num2: Int): Int {
    delay(500)
    return num1 + num2
}

suspend fun parentFun() : Int {
    val num1 = childFun1()
    val num2 = childFun2()
    val sum = childFun3(num1, num2)
    return sum
}

我们可以看到和同步方式操作一模一样，只不过方法上多了 suspend 关键字而已。

注: 这里的 delay 方法，不会像 Thread.sleep 阻塞当前线程。

二. suspend 原理

上面说过，suspend 不会阻塞当前线程，那么它怎么将异步操作的数据，同步传递回来呢？答案其实也是回调，只不过隐藏的很深，我们慢慢分析。

suspend fun childFun1(): Int {
    Thread.sleep(1000)
    return 10
}

suspend fun childFun2(): Int {
    Thread.sleep(2000)
    return 20
}

suspend fun childFun3(num1: Int, num2: Int): Int {
    Thread.sleep(500)
    return num1 + num2
}

suspend fun parentFun() : Int {
    val num1 = childFun1()
    val num2 = childFun2()
    val sum = childFun3(num1, num2)
    return sum
}

这里将 delay 换成 Thread.sleep ,先看 suspend 反编译的 java 代码

public final class CoroutineKt {
   @Nullable
   public static final Object childFun1(@NotNull Continuation $completion) {
      Thread.sleep(1000L);
      return Boxing.boxInt(10);
   }

   @Nullable
   public static final Object childFun2(@NotNull Continuation $completion) {
      Thread.sleep(2000L);
      return Boxing.boxInt(20);
   }

   @Nullable
   public static final Object childFun3(int num1, int num2, @NotNull Continuation $completion) {
      Thread.sleep(500L);
      return Boxing.boxInt(num1 + num2);
   }

   @Nullable
   public static final Object parentFun(@NotNull Continuation $completion) {
      Object $continuation;
      label37: {
         if ($completion instanceof <undefinedtype>) {
            $continuation = (<undefinedtype>)$completion;
            if ((((<undefinedtype>)$continuation).label & Integer.MIN_VALUE) != 0) {
               ((<undefinedtype>)$continuation).label -= Integer.MIN_VALUE;
               break label37;
            }
         }

         $continuation = new ContinuationImpl($completion) {
            // $FF: synthetic field
            Object result;
            int label;
            int I$0;
            int I$1;

            @Nullable
            public final Object invokeSuspend(@NotNull Object $result) {
               this.result = $result;
               this.label |= Integer.MIN_VALUE;
               return CoroutineKt.parentFun(this);
            }
         };
      }

      Object var10000;
      label31: {
         int num1;
         int num2;
         Object var6;
         label30: {
            Object $result = ((<undefinedtype>)$continuation).result;
            var6 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
            switch(((<undefinedtype>)$continuation).label) {
            case 0:
               ResultKt.throwOnFailure($result);
               ((<undefinedtype>)$continuation).label = 1;
               var10000 = childFun1((Continuation)$continuation);
               if (var10000 == var6) {
                  return var6;
               }
               break;
            case 1:
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break;
            case 2:
               num1 = ((<undefinedtype>)$continuation).I$0;
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break label30;
            case 3:
               num2 = ((<undefinedtype>)$continuation).I$1;
               num1 = ((<undefinedtype>)$continuation).I$0;
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break label31;
            default:
               throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
            }

            num1 = ((Number)var10000).intValue();
            ((<undefinedtype>)$continuation).I$0 = num1;
            ((<undefinedtype>)$continuation).label = 2;
            var10000 = childFun2((Continuation)$continuation);
            if (var10000 == var6) {
               return var6;
            }
         }

         num2 = ((Number)var10000).intValue();
         ((<undefinedtype>)$continuation).I$0 = num1;
         ((<undefinedtype>)$continuation).I$1 = num2;
         ((<undefinedtype>)$continuation).label = 3;
         var10000 = childFun3(num1, num2, (Continuation)$continuation);
         if (var10000 == var6) {
            return var6;
         }
      }

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

2.1 Continuation

我们注意到 suspend 修饰的方法，转成 java 方法时，会在方法最后面添加上 Continuation 类型的参数：

suspend fun childFun3(num1: Int, num2: Int): Int {
    Thread.sleep(500)
    return num1 + num2
}
//  变成了
@Nullable
public static final Object childFun3(int num1, int num2, @NotNull Continuation $completion) {
      Thread.sleep(500L);
      return Boxing.boxInt(num1 + num2);
}

这个 Continuation 实例非常重要，它是协程能够实现异步回调的关键对象。

/**
 * Interface representing a continuation after a suspension point that returns a value of type `T`.
 */
@SinceKotlin("1.3")
public interface Continuation<in T> {
    /**
     * The context of the coroutine that corresponds to this continuation.
     */
    public val context: CoroutineContext

    /**
     * Resumes the execution of the corresponding coroutine passing a successful or failed [result] as the
     * return value of the last suspension point.
     */
    public fun resumeWith(result: Result<T>)
}
  1. context : 协程的上下文对象，里面储存着协程上下文环境信息，比如父协程，线程协调器等
  2. resumeWith 方法: 唤醒协程继续执行的方法

注: 这里的唤醒，并不是说协程被线程阻塞了

2.2 协程体

观察 childFun1 、 childFun2 和 childFun3 方法，除了参数上多了一个 $completion 参数，并没有其他变化，那是因为这三个方法中，没有调用其他 suspend 方法，所以和普通函数没有多大区别。

suspend fun parentFun() : Int {
    val num1 = childFun1()
    val num2 = childFun2()
    val sum = childFun3(num1, num2)
    return sum
}

// 转变成了

 @Nullable
   public static final Object parentFun(@NotNull Continuation $completion) {
      Object $continuation;
      label37: {
         if ($completion instanceof <undefinedtype>) {
            $continuation = (<undefinedtype>)$completion;
            if ((((<undefinedtype>)$continuation).label & Integer.MIN_VALUE) != 0) {
               ((<undefinedtype>)$continuation).label -= Integer.MIN_VALUE;
               break label37;
            }
         }

         $continuation = new ContinuationImpl($completion) {
            // $FF: synthetic field
            Object result;
            int label;
            int I$0;
            int I$1;

            @Nullable
            public final Object invokeSuspend(@NotNull Object $result) {
               this.result = $result;
               this.label |= Integer.MIN_VALUE;
               return CoroutineKt.parentFun(this);
            }
         };
      }

      Object var10000;
      label31: {
         int num1;
         int num2;
         Object var6;
         label30: {
            Object $result = ((<undefinedtype>)$continuation).result;
            var6 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
            switch(((<undefinedtype>)$continuation).label) {
            case 0:
               ResultKt.throwOnFailure($result);
               ((<undefinedtype>)$continuation).label = 1;
               var10000 = childFun1((Continuation)$continuation);
               if (var10000 == var6) {
                  return var6;
               }
               break;
            case 1:
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break;
            case 2:
               num1 = ((<undefinedtype>)$continuation).I$0;
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break label30;
            case 3:
               num2 = ((<undefinedtype>)$continuation).I$1;
               num1 = ((<undefinedtype>)$continuation).I$0;
               ResultKt.throwOnFailure($result);
               var10000 = $result;
               break label31;
            default:
               throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
            }

            num1 = ((Number)var10000).intValue();
            ((<undefinedtype>)$continuation).I$0 = num1;
            ((<undefinedtype>)$continuation).label = 2;
            var10000 = childFun2((Continuation)$continuation);
            if (var10000 == var6) {
               return var6;
            }
         }

         num2 = ((Number)var10000).intValue();
         ((<undefinedtype>)$continuation).I$0 = num1;
         ((<undefinedtype>)$continuation).I$1 = num2;
         ((<undefinedtype>)$continuation).label = 3;
         var10000 = childFun3(num1, num2, (Continuation)$continuation);
         if (var10000 == var6) {
            return var6;
         }
      }

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

我们发现 parentFun 方法转成的 java 代码，比我们想象中的要多，这个就是协程实现的秘密。

方法流程分析:

  1. 创建 $continuation 对象, 它是 ContinuationImpl 类的实例，并且储存了方法参数的 $completion 实例，作用是当本协程体(parentFun 方法) 执行完毕之后，会回调 $completion 实例的 resumeWith 方法，唤醒调用方的协程。

注: 当调用 $continuation 对象的 resumeWith 方法会调用 invokeSuspend 方法，就会再次调用 parentFun 方法。

  1. 当 label 不同的时候，执行的逻辑不同:
    1 . 当 label = 0 时，先将 label 设置成 1 ，并调用 childFun1 方法，参数就是当前协程体 $continuation。如果返回的结果值是 COROUTINE_SUSPENDED ，那么就直接 parentFun 方法退出，实现协程挂起。
    2 . 当 label = 1 时，一定是 childFun1 方法内部通过 $continuation 的 resumeWith 方法回调来的，得到 childFun1 方法异步结果值。我们的例子中，不会走到这一步，因为我们直接返回了结果值。
    3 . 当 label = 2 时, 同上，是 childFun2 方法内部通过 $continuation 的 resumeWith 方法回调来的，得到 childFun2 方法异步结果值。
    4 . 当 label = 3 时, 同上原理。

即: label = 1 表示调用了 childFun1 方法；label = 2 表示调用了 childFun2 方法；label = 3 表示调用了 childFun3 方法。并等待 resumeWith 回调带来的结果值。

这种实现方式，我们称之为状态机。

2.3 协程函数的原理

通过上面的分析，我们了解到协程是如何实现挂起和恢复的。
不同于线程的阻塞和唤醒，协程的挂起是方法直接返回，不执行接下来的代码，它的恢复是通过被调用放来实现的。
以上面的例子为例：

  1. parentFun 方法先调用了 childFun1 协程方法，并将自己的协程体 $continuation 传递给 childFun1 方法
  2. 如果 childFun1 方法返回值是 COROUTINE_SUSPENDED ，那么 parentFun 方法直接退出，不会执行 childFun2 方法；直到 childFun1 方法调用 $continuation 的 resumeWith 方法，就会重新调用 parentFun 方法，并因为 label = 1，直接获取结果值，再调用 childFun2 方法。
  3. 如果 childFun1 方法返回值不是 COROUTINE_SUSPENDED，即不需要挂起，那么就继续调用 childFun2 方法。
  4. childFun2 和 childFun3 方法调用过程和 childFun1 方法逻辑一样。

因此我们可以总结:

  1. 当在协程函数内部调用其他协程函数时，都会生成一个挂起点，即这里的 childFun1 childFun2 和 childFun3 ,都对应一个 label 。
  2. 调用其他协程函数的时候，都会将本协程函数的协程体 $continuation 传递给被调用的协程函数，以便被调用的协程函数可以回调恢复本协程函数。
  3. 当被调用的协程函数返回 COROUTINE_SUSPENDED，即表明被调用的协程函数是一个异步操作，希望本协程函数挂起，等待被调用的协程函数执行完成回调它。本协程函数就会从这个挂起点直接返回，不再执行下面的代码，直到被调用的协程函数通过 $continuation 的 resumeWith 方法来恢复本协程函数，并继续执行下面代码，直到遇到下一个挂起点。

本质上协程也是通过回调实现异步操作的，只不过 kotlin 编译器将协程函数变成状态机。
也明白了为什么 suspend 函数为什么只能在 suspend 函数内部调用，而不能在普通函数内部执行，因为没有隐藏的 $continuation 对象。

三. 创建协程

上面分析了 suspend 函数，但是现在这个函数，没办法执行，因为 suspend 函数都需要 Continuation 实例，那么第一个 Continuation 实例该如何创建呢？
kotlin 标准库中提供了两个函数来创建 Continuation 实例

3.1 createCoroutine 方法

@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <T> (suspend () -> T).createCoroutine(
    completion: Continuation<T>
): Continuation<Unit> =
    SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)

@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <R, T> (suspend R.() -> T).createCoroutine(
    receiver: R,
    completion: Continuation<T>
): Continuation<Unit> =
    SafeContinuation(createCoroutineUnintercepted(receiver, completion).intercepted(), COROUTINE_SUSPENDED)

通过 suspend 函数去创建 Continuation 对象，返回 SafeContinuation 类的实例，这个类以后我们有机会分析。
completion 当协程完成之后，会调用它的 resumeWith 方法。

例如:

fun main() {
    val coroutine = (::parentFun).createCoroutine(object : Continuation<Int>{
        override val context: CoroutineContext
            get() = EmptyCoroutineContext

        override fun resumeWith(result: Result<Int>) {
            println("result:$result")
        }
    })
    // 执行
    coroutine.resume(Unit)
}

3.2 startCoroutine 方法

@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

@SinceKotlin("1.3")
@Suppress("UNCHECKED_CAST")
public fun <R, T> (suspend R.() -> T).startCoroutine(
    receiver: R,
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(receiver, completion).intercepted().resume(Unit)
}

这个方法不仅创建了 Continuation 对象，并且还直接执行了协程函数。
例如:

fun main() {
    (::parentFun).startCoroutine(object : Continuation<Int>{
        override val context: CoroutineContext
            get() = EmptyCoroutineContext

        override fun resumeWith(result: Result<Int>) {
            println("result:$result")
        }
    })
}

3.3 suspendCoroutine 方法

上面的例子中 childFun1 这些方法，我们并没有实现异步操作，这里有两个难点:

  1. 没有办法返回 COROUTINE_SUSPENDED 值，因为它不是 Int 类型，且外部获取不到这个值。
  2. 获取不到调用方的 Continuation 对象，来恢复调用方。

针对这种情况，kotlin 提供了 suspendCoroutine 方法来解决这个问题。

@SinceKotlin("1.3")
@InlineOnly
public suspend inline fun <T> suspendCoroutine(crossinline block: (Continuation<T>) -> Unit): T =
    suspendCoroutineUninterceptedOrReturn { c: Continuation<T> ->
        val safe = SafeContinuation(c.intercepted())
        block(safe)
        safe.getOrThrow()
    }

将 childFun1 方法进行改变

suspend fun childFun1(): Int {
    Thread.sleep(1000)
    return 10
}

// 转变成
val executor = Executors.newScheduledThreadPool(1) 

suspend fun childFun1(): Int = suspendCoroutine { continuation ->
    executor.schedule(fun() { continuation.resume(10) }, 1000, TimeUnit.MILLISECONDS)
}

这里我们并没有直接返回，而是通过 executor 线程池，延迟 1 秒钟之后再返回值，模拟异步操作。

转成的 java 代码

   @Nullable
   public static final Object childFun1(@NotNull Continuation $completion) {
      boolean var1 = false;
      boolean var3 = false;
      SafeContinuation var4 = new SafeContinuation(IntrinsicsKt.intercepted($completion));
      Continuation continuation = (Continuation)var4;
      int var6 = false;
      executor.schedule((Runnable)(new CoroutineKt$childFun1$2$1(continuation)), 1000L, TimeUnit.MILLISECONDS);
      Object var10000 = var4.getOrThrow();
      if (var10000 == IntrinsicsKt.getCOROUTINE_SUSPENDED()) {
         DebugProbesKt.probeCoroutineSuspended($completion);
      }

      return var10000;
   }

因为 suspendCoroutine 函数时内联函数，因此函数内容直接复制到 childFun1 函数中。

SafeContinuation 的 getOrThrow 方法

    @PublishedApi
    internal actual fun getOrThrow(): Any? {
        var result = this.result // atomic read
        if (result === UNDECIDED) {
            if (RESULT.compareAndSet(this, UNDECIDED, COROUTINE_SUSPENDED)) return COROUTINE_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
        }
    }

可以看出在没有调用 resumeWith 方法时，就返回 COROUTINE_SUSPENDED

val executor = Executors.newScheduledThreadPool(2)

suspend fun childFun1(): Int = suspendCoroutine { continuation ->
    executor.schedule(fun() { continuation.resume(10) }, 1000, TimeUnit.MILLISECONDS)
}

suspend fun childFun2(): Int = suspendCoroutine { continuation ->
    executor.schedule(fun() { continuation.resume(20) }, 2000, TimeUnit.MILLISECONDS)
}

suspend fun childFun3(num1: Int, num2: Int): Int = suspendCoroutine { continuation ->
    executor.schedule(fun() { continuation.resume(num1 + num2) }, 1000, TimeUnit.MILLISECONDS)
}

suspend fun parentFun() : Int {
    val num1 = childFun1()
    val num2 = childFun2()
    val sum = childFun3(num1, num2)
    return sum
}

fun main() {
    (::parentFun).startCoroutine(object : Continuation<Int>{
        override val context: CoroutineContext
            get() = EmptyCoroutineContext

        override fun resumeWith(result: Result<Int>) {
            println("result:$result")
        }
    })
}
©著作权归作者所有,转载或内容合作请联系作者
  • 人面猴
    序言：七十年代末，一起剥皮案震惊了整个滨河市，随后出现的几起案子，更是在滨河造成了极大的恐慌，老刑警刘岩，带你破解...
    沈念sama阅读 204,793评论 6赞 478
  • 死咒
    序言：滨河连续发生了三起死亡事件，死亡现场离奇诡异，居然都是意外死亡，警方通过查阅死者的电脑和手机，发现死者居然都...
    沈念sama阅读 87,567评论 2赞 381
  • 救了他两次的神仙让他今天三更去死
    文/潘晓璐 我一进店门，熙熙楼的掌柜王于贵愁眉苦脸地迎上来，“玉大人，你说我怎么就摊上这事。” “怎么了？”我有些...
    开封第一讲书人阅读 151,342评论 0赞 338
  • 道士缉凶录：失踪的卖姜人
    文/不坏的土叔 我叫张陵，是天一观的道长。 经常有香客问我，道长，这世上最难降的妖魔是什么？ 我笑而不...
    开封第一讲书人阅读 54,825评论 1赞 277
  • 港岛之恋（遗憾婚礼）
    正文 为了忘掉前任，我火速办了婚礼，结果婚礼上，老公的妹妹穿的比我还像新娘。我一直安慰自己，他们只是感情好，可当我...
    茶点故事阅读 63,814评论 5赞 368
  • 恶毒庶女顶嫁案：这布局不是一般人想出来的
    文/花漫 我一把揭开白布。 她就那样静静地躺着，像睡着了一般。 火红的嫁衣衬着肌肤如雪。 梳的纹丝不乱的头发上，一...
    开封第一讲书人阅读 48,680评论 1赞 281
  • 城市分裂传说
    那天，我揣着相机与录音，去河边找鬼。 笑死，一个胖子当着我的面吹牛，可吹牛的内容都是我干的。 我是一名探鬼主播，决...
    沈念sama阅读 38,033评论 3赞 399
  • 双鸳鸯连环套：你想象不到人心有多黑
    文/苍兰香墨 我猛地睁开眼，长吁一口气：“原来是场噩梦啊……” “哼！你这毒妇竟也来了？” 一声冷哼从身侧响起，我...
    开封第一讲书人阅读 36,687评论 0赞 258
  • 万荣杀人案实录
    序言：老挝万荣一对情侣失踪，失踪者是张志新（化名）和其女友刘颖，没想到半个月后，有当地人在树林里发现了一具尸体，经...
    沈念sama阅读 42,175评论 1赞 300
  • 护林员之死
    正文 独居荒郊野岭守林人离奇死亡，尸身上长有42处带血的脓包…… 初始之章·张勋 以下内容为张勋视角 年9月15日...
    茶点故事阅读 35,668评论 2赞 321
  • 白月光启示录
    正文 我和宋清朗相恋三年，在试婚纱的时候发现自己被绿了。 大学时的朋友给我发了我未婚夫和他白月光在一起吃饭的照片。...
    茶点故事阅读 37,775评论 1赞 332
  • 活死人
    序言：一个原本活蹦乱跳的男人离奇死亡，死状恐怖，灵堂内的尸体忽然破棺而出，到底是诈尸还是另有隐情，我是刑警宁泽，带...
    沈念sama阅读 33,419评论 4赞 321
  • 日本核电站爆炸内幕
    正文 年R本政府宣布，位于F岛的核电站，受9级特大地震影响，放射性物质发生泄漏。R本人自食恶果不足惜，却给世界环境...
    茶点故事阅读 39,020评论 3赞 307
  • 男人毒药：我在死后第九天来索命
    文/蒙蒙 一、第九天 我趴在偏房一处隐蔽的房顶上张望。 院中可真热闹，春花似锦、人声如沸。这庄子的主人今日做“春日...
    开封第一讲书人阅读 29,978评论 0赞 19
  • 一桩弑父案，背后竟有这般阴谋
    文/苍兰香墨 我抬头看了看天上的太阳。三九已至，却和暖如春，着一层夹袄步出监牢的瞬间，已是汗流浃背。 一阵脚步声响...
    开封第一讲书人阅读 31,206评论 1赞 260
  • 情欲美人皮
    我被黑心中介骗来泰国打工， 没想到刚下飞机就差点儿被人妖公主榨干…… 1. 我叫王不留，地道东北人。 一个月前我还...
    沈念sama阅读 45,092评论 2赞 351
  • 代替公主和亲
    正文 我出身青楼，却偏偏与公主长得像，于是被迫代替她去往敌国和亲。 传闻我的和亲对象是个残疾皇子，可洞房花烛夜当晚...
    茶点故事阅读 42,510评论 2赞 343

推荐阅读更多精彩内容