• Kotlin协程 - launch原理 笔记

    一、协程是如何创建的?

    launch、async 可以创建、启动新的协程,那么协程到底是如何创建的？

    1.   runBlocking {
    2.         println(Thread.currentThread().name)
    3.         launch {
    4.             println(Thread.currentThread().name)
    5.             delay(100L)
    6.         }
    7.         Thread.sleep(1000L)
    8.     }
    9.  
    10.  
    11. Log
    12. main @coroutine#1
    13. main @coroutine#2
    14.  
    15. Process finished with exit code 0

     runBlocking{} 启动了第一个协程，launch{} 启动了第二个协程。

    1.协程启动的基础 API

    1.  
    2.  
    3. public fun  (suspend () -> T).createCoroutine(
    4.     completion: Continuation
    5. ): Continuation<Unit> =
    6.     SafeContinuation(createCoroutineUnintercepted(completion).intercepted(), COROUTINE_SUSPENDED)
    7.  
    8. public fun  (suspend () -> T).startCoroutine(
    9.     completion: Continuation
    10. ) {
    11.     createCoroutineUnintercepted(completion).intercepted().resume(Unit)
    12. }

    createCoroutine{}、startCoroutine{}就是 Kotlin 协程当中最基础的两个创建协程的 API。启动协程有三种常见的方式：launch、runBlocking、async。它们其实属于协程中间层提供的 API，而它们的底层都调用了“基础层”的协程 API。

    createCoroutine{}、startCoroutine{}是扩展函数，其扩展接收者类型是一个函数类型：suspend () -> T，代表了“无参数，返回值为 T 的挂起函数或者 Lambda”。而对于函数本身，它们两个都接收一个 Continuation 类型的参数，其中一个函数，还会返回一个 Continuation 类型的返回值。

    1. val block = suspend {
    2.     println("Hello")
    3.     delay(1000L)
    4.     println("World!")
    5.     "Result"
    6. }
    7.  
    8. fun testLaunch2() {
    9.     val continuation = object : Continuation {
    10.         override val context: CoroutineContext
    11.             get() = EmptyCoroutineContext
    12.  
    13.         override fun resumeWith(result: Result<String>) {
    14.             println("Result:" + result.getOrNull())
    15.         }
    16.     }
    17.     block.startCoroutine(continuation)
    18. }
    19.  
    20. fun main() {
    21.     testLaunch2()
    22.     Thread.sleep(2000L)
    23. }
    24.  
    25. Log
    26. Hello
    27. World!
    28. Result:Result
    29.  
    30. Process finished with exit code 0

    类型为 suspend () -> T的函数或者 Lambda 表达式可以用 block.startCoroutine() 来启动协程了。

    Continuation 有两种用法，一种是在实现挂起函数的时候，用于传递挂起函数的执行结果；另一种是在调用挂起函数的时候，以匿名内部类的方式，用于接收挂起函数的执行结果。

    使用 createCoroutine() 这个方法其实上面代码的逻辑：

    1. fun testLaunch3() {
    2.     val continuation = object : Continuation {
    3.         override val context: CoroutineContext
    4.             get() = EmptyCoroutineContext
    5.  
    6.         override fun resumeWith(result: Result<String>) {
    7.             println("Result:" + result.getOrNull())
    8.         }
    9.     }
    10.     val coroutinue = block.createCoroutine(continuation)
    11.     coroutinue.resume(Unit)
    12. }
    13.  
    14.  
    15. val block = suspend {
    16.     println("Hello")
    17.     delay(1000L)
    18.     println("World!")
    19.     "Result"
    20. }
    21.  
    22. fun main() {
    23.     testLaunch3()
    24.     Thread.sleep(2000L)
    25. }
    26.  
    27. Log
    28. Hello
    29. World!
    30. Result:Result
    31.  
    32. Process finished with exit code 0

     createCoroutine() 创建一个协程，先不启动。调用 resume() 才能启动。 createCoroutine()、startCoroutine() 的源代码差别也并不大，只是前者没有调用 resume()，而后者调用了 resume()。startCoroutine() 之所以可以创建并同时启动协程的原因就在于，它在源码中直接调用了 resume(Unit)。

    将 startCoroutine()转换为Java：

    1. package com.example.myapplication.testcoroutinue;
    2.  
    3. import kotlin.Metadata;
    4. import kotlin.Result;
    5. import kotlin.ResultKt;
    6. import kotlin.Unit;
    7. import kotlin.coroutines.Continuation;
    8. import kotlin.coroutines.ContinuationKt;
    9. import kotlin.coroutines.CoroutineContext;
    10. import kotlin.coroutines.EmptyCoroutineContext;
    11. import kotlin.coroutines.intrinsics.IntrinsicsKt;
    12. import kotlin.jvm.functions.Function1;
    13. import kotlin.jvm.internal.Intrinsics;
    14. import kotlinx.coroutines.DelayKt;
    15. import org.jetbrains.annotations.NotNull;
    16. import org.jetbrains.annotations.Nullable;
    17.  
    18. @Metadata(
    19.    mv = {1, 6, 0},
    20.    k = 2,
    21.    d1 = {"\u0000\u001e\n\u0000\n\u0002\u0018\u0002\n\u0002\u0018\u0002\n\u0002\u0010\u000e\n\u0002\u0010\u0000\n\u0002\b\u0004\n\u0002\u0010\u0002\n\u0002\b\u0002\u001a\u0006\u0010\b\u001a\u00020\t\u001a\u0006\u0010\n\u001a\u00020\t\",\u0010\u0000\u001a\u0018\b\u0001\u0012\n\u0012\b\u0012\u0004\u0012\u00020\u00030\u0002\u0012\u0006\u0012\u0004\u0018\u00010\u00040\u0001ø\u0001\u0000¢\u0006\n\n\u0002\u0010\u0007\u001a\u0004\b\u0005\u0010\u0006\u0082\u0002\u0004\n\u0002\b\u0019¨\u0006\u000b"},
    22.    d2 = {"block", "Lkotlin/Function1;", "Lkotlin/coroutines/Continuation;", "", "", "getBlock", "()Lkotlin/jvm/functions/Function1;", "Lkotlin/jvm/functions/Function1;", "main", "", "testLaunch2", "My_Application.app.main"}
    23. )
    24. public final class TestCoroutinue888Kt {
    25. // Kotlin 为 block 变量生成的静态变量
    26.    @NotNull
    27.    private static final Function1 block;
    28.  
    29.    public static final void main() {
    30.       testLaunch2();
    31.       Thread.sleep(2000L);
    32.    }
    33.  
    34.    // $FF: synthetic method
    35.    public static void main(String[] var0) {
    36.       main();
    37.    }
    38.     
    39.     // Kotlin 为 block 变量生成的静态变量以及方法
    40.  
    41.    @NotNull
    42.    public static final Function1 getBlock() {
    43.       return block;
    44.    }
    45.  
    46.    public static final void testLaunch2() {
    47. //continuation 变量对应的匿名内部类
    48.        continuation = new Continuation() {
    49.          @NotNull
    50.          public CoroutineContext getContext() {
    51.             return (CoroutineContext)EmptyCoroutineContext.INSTANCE;
    52.          }
    53.  
    54.          public void resumeWith(@NotNull Object result) {
    55.             String var2 = "Result:" + (String)(Result.isFailure-impl(result) ? null : result);
    56.             System.out.println(var2);
    57.          }
    58.       };
    59. //block.startCoroutine(continuation) 转换成了ContinuationKt.startCoroutine(block, (Continuation)continuation)
    60.       ContinuationKt.startCoroutine(block, (Continuation)continuation);
    61.    }
    62.  
    63.    static {
    64.     //实现了 Continuation 接口
    65.       Function1 var0 = (Function1)(new Function1((Continuation)null) {
    66.          int label;
    67. //invokeSuspend()为协程状态机逻辑
    68.  
    69.          @Nullable
    70.          public final Object invokeSuspend(@NotNull Object $result) {
    71.             Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    72.             String var2;
    73.             switch(this.label) {
    74.             case 0:
    75.                ResultKt.throwOnFailure($result);
    76.                var2 = "Hello";
    77.                System.out.println(var2);
    78.                this.label = 1;
    79.                if (DelayKt.delay(1000L, this) == var3) {
    80.                   return var3;
    81.                }
    82.                break;
    83.             case 1:
    84.                ResultKt.throwOnFailure($result);
    85.                break;
    86.             default:
    87.                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
    88.             }
    89.  
    90.             var2 = "World!";
    91.             System.out.println(var2);
    92.             return "Result";
    93.          }
    94.  
    95.          @NotNull
    96.          public final Continuation create(@NotNull Continuation completion) {
    97.             Intrinsics.checkNotNullParameter(completion, "completion");
    98.             Function1 var2 = new (completion);
    99.             return var2;
    100.          }
    101.  
    102.          public final Object invoke(Object var1) {
    103.             return (()this.create((Continuation)var1)).invokeSuspend(Unit.INSTANCE);
    104.          }
    105.       });
    106.       block = var0;
    107.    }
    108. }

     

    1.  
    2.  
    3.  
    4. public fun  (suspend () -> T).startCoroutine(
    5.     completion: Continuation
    6. ) {
    7. createCoroutineUnintercepted(completion).intercepted().resume(Unit)
    8. }

    在 startCoroutine() 当中，首先会调用 createCoroutineUnintercepted() 方法。

    1.  
    2. public expect fun  (suspend () -> T).createCoroutineUnintercepted(
    3.     completion: Continuation
    4. ): Continuation<Unit>

     代码中的 expect，一种声明，由于 Kotlin 是面向多个平台的，具体的实现，就需要在特定的平台实现。

    1.  
    2. public actual fun  (suspend () -> T).createCoroutineUnintercepted(
    3.     completion: Continuation
    4. ): Continuation<Unit> {
    5.     val probeCompletion = probeCoroutineCreated(completion)
    6.     return if (this is BaseContinuationImpl)
    7.         create(probeCompletion)
    8.     else
    9.         createCoroutineFromSuspendFunction(probeCompletion) {
    10.             (this as Function1, Any?>).invoke(it)
    11.         }
    12. }

      actual，代表了 createCoroutineUnintercepted() 在 JVM 平台的实现。

    createCoroutineUnintercepted() 是一个扩展函数，this代表了 block 变量。(this is BaseContinuationImpl) 条件为ture，就会调用 create(probeCompletion)。

    1.  
    2.  
    3.  
    4. public open fun create(completion: Continuation<*>): Continuation<Unit> {
    5.     throw UnsupportedOperationException("create(Continuation) has not been overridden")
    6. }

    在默认情况下，这个 create() 方法是会抛出异常的。

    1.  
    2.          @NotNull
    3.          public final Continuation create(@NotNull Continuation completion) {
    4.             Intrinsics.checkNotNullParameter(completion, "completion");
    5.             Function1 var2 = new (completion);
    6.             return var2;
    7.          }

    返回了Continuation 对象。

    1.  
    2.  
    3. public fun  (suspend () -> T).startCoroutine(
    4.     completion: Continuation
    5. ) {
    6.  
    7. createCoroutineUnintercepted(completion).intercepted().resume(Unit)
    8. }

     intercepted() 在 JVM 实现如下：

    1.  
    2.  
    3.  
    4. public actual fun  Continuation.intercepted(): Continuation =
    5.     (this as? ContinuationImpl)?.intercepted() ?: this

    将 Continuation 强转成了 ContinuationImpl，调用了它的 intercepted()。

    ContinuationImpl 的源代码:

    1.  
    2.  
    3. internal abstract class ContinuationImpl(
    4.     completion: Continuation?,
    5.     private val _context: CoroutineContext?
    6. ) : BaseContinuationImpl(completion) {
    7.  
    8.     @Transient
    9.     private var intercepted: Continuation? = null
    10.  
    11.     public fun intercepted(): Continuation =
    12.         intercepted
    13.             ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
    14.                 .also { intercepted = it }
    15. }

     

    通过 ContinuationInterceptor，对 Continuation 进行拦截，从而将程序的执行逻辑派发到特定的线程之上。

     resume(Unit)：

    1.  
    2.  
    3. public fun  (suspend () -> T).startCoroutine(
    4.     completion: Continuation
    5. ) {
    6.                                                      
    7. createCoroutineUnintercepted(completion).intercepted().resume(Unit)
    8. }

      resume(Unit)，作用其实就相当于启动了协程。

    二、launch 是如何启动协程的？

    1. fun main() {
    2.     testLaunch11()
    3.     Thread.sleep(2000L)
    4. }
    5.  
    6. fun testLaunch11() {
    7.     val coroutineScope = CoroutineScope(Job())
    8.     coroutineScope.launch {
    9.         println("Hello")
    10.         delay(1000L)
    11.         println("World!")
    12.     }
    13. }
    14.  
    15. Log
    16. Hello
    17. World!
    18.  
    19. Process finished with exit code 0

    转Java

    1. package com.example.myapplication.testcoroutinue;
    2.  
    3. import kotlin.Metadata;
    4. import kotlin.ResultKt;
    5. import kotlin.Unit;
    6. import kotlin.coroutines.Continuation;
    7. import kotlin.coroutines.CoroutineContext;
    8. import kotlin.coroutines.intrinsics.IntrinsicsKt;
    9. import kotlin.jvm.functions.Function2;
    10. import kotlin.jvm.internal.Intrinsics;
    11. import kotlinx.coroutines.BuildersKt;
    12. import kotlinx.coroutines.CoroutineScope;
    13. import kotlinx.coroutines.CoroutineScopeKt;
    14. import kotlinx.coroutines.CoroutineStart;
    15. import kotlinx.coroutines.DelayKt;
    16. import kotlinx.coroutines.Job;
    17. import kotlinx.coroutines.JobKt;
    18. import org.jetbrains.annotations.NotNull;
    19. import org.jetbrains.annotations.Nullable;
    20.  
    21. @Metadata(
    22.    mv = {1, 6, 0},
    23.    k = 2,
    24.    d1 = {"\u0000\n\n\u0000\n\u0002\u0010\u0002\n\u0002\b\u0002\u001a\u0006\u0010\u0000\u001a\u00020\u0001\u001a\u0006\u0010\u0002\u001a\u00020\u0001¨\u0006\u0003"},
    25.    d2 = {"main", "", "testLaunch11", "My_Application.app.main"}
    26. )
    27. public final class TestCoroutinue999Kt {
    28.    public static final void main() {
    29.       testLaunch11();
    30.       Thread.sleep(2000L);
    31.    }
    32.  
    33.    // $FF: synthetic method
    34.    public static void main(String[] var0) {
    35.       main();
    36.    }
    37.  
    38.    public static final void testLaunch11() {
    39.       CoroutineScope coroutineScope = CoroutineScopeKt.CoroutineScope((CoroutineContext)JobKt.Job$default((Job)null, 1, (Object)null));
    40. //对应 launch 当中的 Lambda。
    41.       BuildersKt.launch$default(coroutineScope, (CoroutineContext)null, (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
    42.          int label;
    43.  
    44.          @Nullable
    45.          public final Object invokeSuspend(@NotNull Object $result) {
    46.             Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
    47.             String var2;
    48.             switch(this.label) {
    49.             case 0:
    50.                ResultKt.throwOnFailure($result);
    51.                var2 = "Hello";
    52.                System.out.println(var2);
    53.                this.label = 1;
    54.                if (DelayKt.delay(1000L, this) == var3) {
    55.                   return var3;
    56.                }
    57.                break;
    58.             case 1:
    59.                ResultKt.throwOnFailure($result);
    60.                break;
    61.             default:
    62.                throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
    63.             }
    64.  
    65.             var2 = "World!";
    66.             System.out.println(var2);
    67.             return Unit.INSTANCE;
    68.          }
    69.  
    70.          @NotNull
    71.          public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
    72.             Intrinsics.checkNotNullParameter(completion, "completion");
    73.             Function2 var3 = new (completion);
    74.             return var3;
    75.          }
    76.  
    77.          public final Object invoke(Object var1, Object var2) {
    78.             return (()this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
    79.          }
    80.       }), 3, (Object)null);
    81.    }
    82. }

    launch源码

    1.  
    2. public fun CoroutineScope.launch(
    3.     context: CoroutineContext = EmptyCoroutineContext,
    4.     start: CoroutineStart = CoroutineStart.DEFAULT,
    5.     block: suspend CoroutineScope.() -> Unit
    6. ): Job {
    7.     //launch 会根据传入的 CoroutineContext 创建出新的 Context。
    8.     val newContext = newCoroutineContext(context)
    9.     //launch 会根据传入的启动模式来创建对应的协程对象。这里有两种，一种是标准的，一种是懒加载的。
    10.     val coroutine = if (start.isLazy)
    11.         LazyStandaloneCoroutine(newContext, block) else
    12.         StandaloneCoroutine(newContext, active = true)
    13.     //启动协程。
    14.     coroutine.start(start, coroutine, block)
    15.     return coroutine
    16. }

      coroutine.start() :

    1.  
    2. public abstract class AbstractCoroutine<in T>(
    3.     parentContext: CoroutineContext,
    4.     initParentJob: Boolean,
    5.     active: Boolean
    6. ) : JobSupport(active), Job, Continuation, CoroutineScope {
    7.  
    8.    
    9.  
    10.     public fun  start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    11.         start(block, receiver, this)
    12.     }
    13. }

    AbstractCoroutine.kt 对应协程的抽象逻辑。AbstractCoroutine 的 start() 方法，用于启动协程。 

    1.  
    2. public enum class CoroutineStart {
    3.     public operator fun  invoke(block: suspend () -> T, completion: Continuation<T>): Unit =
    4.         when (this) {
    5.             DEFAULT -> block.startCoroutineCancellable(completion)
    6.             ATOMIC -> block.startCoroutine(completion)
    7.             UNDISPATCHED -> block.startCoroutineUndispatched(completion)
    8.             LAZY -> Unit // will start lazily
    9.         }
    10. }

    start(block, receiver, this)，进入 CoroutineStart.invoke()。

     invoke() 方法当中，根据 launch 传入的启动模式，以不同的方式启动协程。当启动模式是 ATOMIC 的时候，就会调用 block.startCoroutine(completion)。startCoroutineUndispatched(completion) 和 startCoroutineCancellable(completion)，只是在 startCoroutine() 的基础上增加了一些额外的功能而已。前者代表启动协程以后就不会被分发，后者代表启动以后可以响应取消。

    startCoroutineCancellable(completion)

    1.  
    2. public fun  (suspend () -> T).startCoroutineCancellable(completion: Continuation): Unit = runSafely(completion) {
    3.     
    4.     createCoroutineUnintercepted(completion).intercepted().resumeCancellableWith(Result.success(Unit))
    5. }
    6.  
    7. public actual fun  (suspend () -> T).createCoroutineUnintercepted(
    8.     completion: Continuation
    9. ): Continuation<Unit> {
    10.     val probeCompletion = probeCoroutineCreated(completion)
    11.  
    12.     return if (this is BaseContinuationImpl)
    13.         
    14.         create(probeCompletion)
    15.     else
    16.         createCoroutineFromSuspendFunction(probeCompletion) {
    17.             (this as Function1, Any?>).invoke(it)
    18.         }
    19. }

    startCoroutineCancellable() 的源代码，会调用 createCoroutineUnintercepted()，然后调用 create(probeCompletion)，然后最终会调用create() 方法。launch 这个 API，只是对协程的基础元素 startCoroutine() 等方法进行了一些封装而已。

  • 相关阅读:
    若依vue中字典Dict插件的研究
    利用PyTorch训练模型识别数字+英文图片验证码
    做SaaS的程序员们，是时候关注企业架构了
    java计算机毕业设计进出货管理系统源码+mysql数据库+系统+lw文档+部署
    NSTextField如何实现字体居中
    【开源】历史学习网站 JAVA+Vue.js+SpringBoot+MySQL
    万界星空科技MES系统中的车间管理的作用
    Himall商城Web帮助类获得请求客户端的操作系统名称、判断是否是浏览器请求、是否是移动设备请求、判断是否是搜索引擎爬虫请求
    java牛客笔试题练习日记 - 集合篇
    微信小程序获取用户信息
  • 原文地址：https://blog.csdn.net/zhangying1994/article/details/127880236