Kotlin协程launch启动流程原理详解

目录
  • 1.launch启动流程
    • 反编译后的Java代码
  • 2.协程是如何被启动的

1.launch启动流程

已知协程的启动方式之一是Globalscope.launch,那么Globalscope.launch的流程是怎样的呢,直接进入launch的源码开始看起。

fun main() {
    coroutineTest()
    Thread.sleep(2000L)
}
val block = suspend {
    println("Hello")
    delay(1000L)
    println("Kotlin")
}
private fun coroutineTest() {
    CoroutineScope(Job()).launch {
        withContext(Dispatchers.IO) {
            block.invoke()
        }
    }
}

反编译后的Java代码

public final class CoroutineDemoKt {
   @NotNull
   private static final Function1 block;
   public static final void main() {
      coroutineTest();
      Thread.sleep(2000L);
   }
   // $FF: synthetic method
   public static void main(String[] var0) {
      main();
   }
   @NotNull
   public static final Function1 getBlock() {
      return block;
   }
   private static final void coroutineTest() {
      BuildersKt.launch$default(CoroutineScopeKt.CoroutineScope((CoroutineContext)JobKt.Job$default((Job)null, 1, (Object)null)), (CoroutineContext)null, (CoroutineStart)null, (Function2)(new Function2((Continuation)null) {
         int label;
         @Nullable
         public final Object invokeSuspend(@NotNull Object $result) {
            Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
            switch(this.label) {
            case 0:
               ResultKt.throwOnFailure($result);
               CoroutineContext var10000 = (CoroutineContext)Dispatchers.getIO();
               Function2 var10001 = (Function2)(new Function2((Continuation)null) {
                  int label;
                  @Nullable
                  public final Object invokeSuspend(@NotNull Object $result) {
                     Object var2 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
                     switch(this.label) {
                     case 0:
                        ResultKt.throwOnFailure($result);
                        Function1 var10000 = CoroutineDemoKt.getBlock();
                        this.label = 1;
                        if (var10000.invoke(this) == var2) {
                           return var2;
                        }
                        break;
                     case 1:
                        ResultKt.throwOnFailure($result);
                        break;
                     default:
                        throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
                     }
                     return Unit.INSTANCE;
                  }
                  @NotNull
                  public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
                     Intrinsics.checkNotNullParameter(completion, "completion");
                     Function2 var3 = new <anonymous constructor>(completion);
                     return var3;
                  }
                  public final Object invoke(Object var1, Object var2) {
                     return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
                  }
               });
               this.label = 1;
               if (BuildersKt.withContext(var10000, var10001, this) == var2) {
                  return var2;
               }
               break;
            case 1:
               ResultKt.throwOnFailure($result);
               break;
            default:
               throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
            }
            return Unit.INSTANCE;
         }
         @NotNull
         public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
            Intrinsics.checkNotNullParameter(completion, "completion");
            Function2 var3 = new <anonymous constructor>(completion);
            return var3;
         }
         public final Object invoke(Object var1, Object var2) {
            return ((<undefinedtype>)this.create(var1, (Continuation)var2)).invokeSuspend(Unit.INSTANCE);
         }
      }), 3, (Object)null);
   }
   static {
      Function1 var0 = (Function1)(new Function1((Continuation)null) {
         int label;
         @Nullable
         public final Object invokeSuspend(@NotNull Object $result) {
            Object var3 = IntrinsicsKt.getCOROUTINE_SUSPENDED();
            String var2;
            switch(this.label) {
            case 0:
               ResultKt.throwOnFailure($result);
               var2 = "Hello";
               System.out.println(var2);
               this.label = 1;
               if (DelayKt.delay(1000L, this) == var3) {
                  return var3;
               }
               break;
            case 1:
               ResultKt.throwOnFailure($result);
               break;
            default:
               throw new IllegalStateException("call to 'resume' before 'invoke' with coroutine");
            }
            var2 = "Kotlin";
            System.out.println(var2);
            return Unit.INSTANCE;
         }
         @NotNull
         public final Continuation create(@NotNull Continuation completion) {
            Intrinsics.checkNotNullParameter(completion, "completion");
            Function1 var2 = new <anonymous constructor>(completion);
            return var2;
         }
         public final Object invoke(Object var1) {
            return ((<undefinedtype>)this.create((Continuation)var1)).invokeSuspend(Unit.INSTANCE);
         }
      });
      block = var0;
   }
}

先分析一下上面代码的流程:

  • 首先声明了一个Function1类型的block变量,这个变量就是demo中的block,然后会在static函数中会被赋值。
  • 接下来就是coroutineTest函数的调用。这个函数中的第一行代码就是CoroutineScope的传参和一些默认值
  • 然后通过89行的invoke进入到了外层状态机流转的过程
  • 95行的static表示的是内部的挂起函数就是demo中的block.invoke,它是以匿名内部类的方式实现,然后执行内部的状态机流转过程,最后给block赋值。
  • block被赋值后最终在Function1 var10000 = CoroutineDemoKt.getBlock();被调用

那么这个过程又是如何实现的,进入launch源码进行查看:

public fun CoroutineScope.launch(
    context: CoroutineContext = EmptyCoroutineContext,
    start: CoroutineStart = CoroutineStart.DEFAULT,
    block: suspend CoroutineScope.() -> Unit
): Job {
    val newContext = newCoroutineContext(context)
    val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    coroutine.start(start, coroutine, block)
    return coroutine
}

这里的block指的就是demo中的block代码段

再来看一下里面的几行代码的含义:

  • newCoroutineContext: 通过默认的或者传入的context创建一个新的Context;
  • coroutine: launch 会根据传入的启动模式来创建对应的协程对象。这里有两种,一种是标准的,一种是懒加载的。
  • coroutine.start: 尝试启动协程

2.协程是如何被启动的

通过launch的源码可知协程的启动是通过coroutine.start启动的,那么协程的启动流程又是怎样的?

public abstract class AbstractCoroutine<in T>(
    parentContext: CoroutineContext,
    initParentJob: Boolean,
    active: Boolean
) : JobSupport(active), Job, Continuation<T>, CoroutineScope {
    ...
    /**
     * 用给定的代码块启动这个协程并启动策略。这个函数在这个协程上最多调用一次。
     */
    public fun <R> start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
        start(block, receiver, this)
    }
}

start函数中传入了三个参数,只需要关注第一个参数即可。

public enum class CoroutineStart {
    ...
    /**
     * 用这个协程的启动策略启动相应的块作为协程。
     */
    public operator fun <T> invoke(block: suspend () -> T, completion: Continuation<T>): Unit =
    when (this) {
        DEFAULT -> block.startCoroutineCancellable(completion)
        ATOMIC -> block.startCoroutine(completion)
        UNDISPATCHED -> block.startCoroutineUndispatched(completion)
        LAZY -> Unit // will start lazily
    }
}

启动策略的具体实现有三种方式,这里只需要分析startCoroutine,另外两个其实就是它的基础上增加了一些功能,其中前者代表启动协程以后可以在等待调度时取消,后者表示协程启动后不会被分发。

/**
 * 创建没有接收方且结果类型为T的协程,这个函数每次调用时都会创建一个新的可挂起的实例。
 */
public fun <T> (suspend () -> T).startCoroutine(
    completion: Continuation<T>
) {
    createCoroutineUnintercepted(completion).intercepted().resume(Unit)
}

createCoroutineUnintercepted在源代码中只是一个声明,它的具体实现是在IntrinsicsJvm.kt文件中。

//IntrinsicsJvm.kt#createCoroutineUnintercepted
/**
 * 创建没有接收方且结果类型为T的非拦截协程。这个函数每次调用时都会创建一个新的可挂起的实例。
 */
public actual fun <T> (suspend () -> T).createCoroutineUnintercepted(
    completion: Continuation<T>
): Continuation<Unit> {
    val probeCompletion = probeCoroutineCreated(completion)
    return if (this is BaseContinuationImpl)
        create(probeCompletion)
    else
        createCoroutineFromSuspendFunction(probeCompletion) {
            (this as Function1<Continuation<T>, Any?>).invoke(it)
        }
}

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

createCoroutineUnintercepted是一个扩展函数,接收者类型是一个无参数,返回值为 T 的挂起函数或者 Lambda。

第9行代码中的this代表的是(suspend () -> T)也就是invoke函数中的block变量,这个block变量就是demo中的block代码段。

第9行的BaseContinuationImpl是一个抽象类它实现了Continuation

关于if (this is BaseContinuationImpl)的结果暂且不分析,先分析两种情况下的create函数:

  • create(probeCompletion):
//ContinuationImpl.kt#create
public open fun create(completion: Continuation<*>): Continuation<Unit> {
    throw UnsupportedOperationException("create(Continuation) has not been overridden")
}
public open fun create(value: Any?, completion: Continuation<*>): Continuation<Unit> {
    throw UnsupportedOperationException("create(Any?;Continuation) has not been overridden")
}

这个create函数抛出一个异常,意思就是这个create()没有被重写,而这个create()的重写就是在反编译后的Java代码中的create函数

@NotNull
public final Continuation create(@Nullable Object value, @NotNull Continuation completion) {
    Intrinsics.checkNotNullParameter(completion, "completion");
    Function2 var3 = new <anonymous constructor>(completion);
    return var3;
}
  • createCoroutineFromSuspendFunction(probeCompletion):
//IntrinsicsJvm.kt#createCoroutineFromSuspendFunction
/**
 * 当一个被suspend修饰的lambda表达式没有继承BaseContinuationImpl类时,则通过此方法创建协程。
 *
 * 它发生在两种情况下:
 * 1.lambda表达式中调用了其他的挂起方法
 * 2.挂起方法是通过Java实现的
 *
 * 必须将它封装到一个扩展[BaseContinuationImpl]的实例中,因为这是所有协程机制的期望。
 */
private inline fun <T> createCoroutineFromSuspendFunction(
	completion: Continuation<T>,
	crossinline block: (Continuation<T>) -> Any?
		): Continuation<Unit> {
	val context = completion.context
	// context为空创建一个受限协程
	return if (context === EmptyCoroutineContext)
	//受限协程:只能调用协程作用域中提供的挂起方式挂起,其他挂起方法不能调用
	object : RestrictedContinuationImpl(completion as Continuation<Any?>) {
		private var label = 0
		override fun invokeSuspend(result: Result<Any?>): Any? =
		when (label) {
			0 -> {
				label = 1
				result.getOrThrow() // 如果试图以异常开始,则重新抛出异常(将被BaseContinuationImpl.resumeWith捕获)
				block(this) // 运行块,可以返回或挂起
			}
			1 -> {
				label = 2
				result.getOrThrow() // 这是block挂起的结果
			}
			else -> error("This coroutine had already completed")
		}
	}
	else
	//创建一个正常的协程
	object : ContinuationImpl(completion as Continuation<Any?>, context) {
		private var label = 0
		override fun invokeSuspend(result: Result<Any?>): Any? =
		when (label) {
			0 -> {
				label = 1
				result.getOrThrow() // 如果试图以异常开始,则重新抛出异常(将被BaseContinuationImpl.resumeWith捕获)
				block(this) // 运行块,可以返回或挂起
			}
			1 -> {
				label = 2
				result.getOrThrow() // 这是block挂起的结果
			}
			else -> error("This coroutine had already completed")
		}
	}
}

createCoroutineFromSuspendFunction就是当一个被suspend修饰的Lambda表达式没有继承BaseContinuationImpl是才会被调用,然后根据上下文是否为空创建不同类型的协程。

两种情况都已经分析完了,那么现在if (this is BaseContinuationImpl)会执行哪一个呢,首先这里的this所指的就是demo中的block代码段,Kotlin编译器编译后会自动生成一个类就是上面的static,它会继承SuspendLambda类,而这个SuspendLambda类继承自ContinuationImpl,ContinuationImpl继承自BaseContinuationImpl,因此可以得到判断结果为true,

createCoroutineUnintercepted的过程就是协程创建的过程。

然后就是intercepted函数,这个函数的具体实现也在IntrinsicsJvm.kt中,那么intercepted又做了什么呢

public expect fun <T> Continuation<T>.intercepted(): Continuation<T>
//具体实现
//IntrinsicsJvm.kt#intercepted
public actual fun <T> Continuation<T>.intercepted(): Continuation<T> =
    (this as? ContinuationImpl)?.intercepted() ?: this

首先有个强转,通过上面的分析这个强转是一定会成功的,到这里intercepted就进入到了ContinuationImpl中了

internal abstract class ContinuationImpl(
    completion: Continuation<Any?>?,
    private val _context: CoroutineContext?
) : BaseContinuationImpl(completion) {
	...
    @Transient
    private var intercepted: Continuation<Any?>? = null
	//如果没有缓存,则从上下文获取拦截器,调用interceptContinuation进行拦截
	//将获取到的内容保存到全局变量
    public fun intercepted(): Continuation<Any?> =
        intercepted
            ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
                .also { intercepted = it }
}

这里的ContinuationInterceptor指的就是Demo中传输的Dispatcher.IO,默认值时Dispatcher.Default

再回到startContinue中还剩最后一个resume

/**
 * 恢复执行相应的协程传递值作为最后一个挂起点的返回值。
 */
public inline fun <T> Continuation<T>.resume(value: T): Unit =
	resumeWith(Result.success(value))
public interface Continuation<in T> {
	/**
     * 与此延续相对应的协程的上下文。
     */
	public val context: CoroutineContext
	/**
     * 恢复执行相应的协程传递值作为最后一个挂起点的返回值。
     */
	public fun resumeWith(result: Result<T>)
}

这里的resume(Unit)作用就相当与启动了一个协程。

上面的启动流程中为了方便分析的是CoroutineStart.ATOMIC,而默认的是CoroutineStart.DEFAULT,下面分析一下DEFAULT的流程

//Cancellable.kt#startCoroutineCancellable
public fun <T> (suspend () -> T).startCoroutineCancellable(completion: Continuation<T>): Unit = runSafely(completion) {
    createCoroutineUnintercepted(completion).intercepted().resumeCancellableWith(Result.success(Unit))
}

startCoroutineCancellable对于协程的创建和拦截与ATOMIC是一样的,区别就在于resumeCancellableWith

//DispatchedContinuation#resumeCancellableWith
public fun <T> Continuation<T>.resumeCancellableWith(
	result: Result<T>,
	onCancellation: ((cause: Throwable) -> Unit)? = null
): Unit = when (this) {
	is DispatchedContinuation -> resumeCancellableWith(result, onCancellation)
	else -> resumeWith(result)
}
// 我们内联它来保存堆栈上的一个条目,在它显示的情况下(无限制调度程序)
// 它只在Continuation<T>.resumeCancellableWith中使用
@Suppress("NOTHING_TO_INLINE")
inline fun resumeCancellableWith(
	result: Result<T>,
	noinline onCancellation: ((cause: Throwable) -> Unit)?
		) {
	val state = result.toState(onCancellation)
	//是否需要分发
	if (dispatcher.isDispatchNeeded(context)) {
		_state = state
		resumeMode = MODE_CANCELLABLE
		//将可运行块的执行分派给给定上下文中的另一个线程
		dispatcher.dispatch(context, this)
	} else {
		executeUnconfined(state, MODE_CANCELLABLE) {
			//协程未被取消
			if (!resumeCancelled(state)) {
				// 恢复执行
				resumeUndispatchedWith(result)
			}
		}
	}
}
//恢复执行前判断协程是否已经取消执行
inline fun resumeCancelled(state: Any?): Boolean {
	//获取当前协程任务
	val job = context[Job]
	//如果不为空且不活跃
	if (job != null && !job.isActive) {
		val cause = job.getCancellationException()
		cancelCompletedResult(state, cause)
		//抛出异常
		resumeWithException(cause)
		return true
	}
	return false
}
//我们需要内联它来在堆栈中保存一个条目
inline fun resumeUndispatchedWith(result: Result<T>) {
	withContinuationContext(continuation, countOrElement) {
		continuation.resumeWith(result)
	}
}

以上就是Kotlin协程launch启动流程原理详解的详细内容,更多关于Kotlin协程launch启动流程的资料请关注我们其它相关文章!

(0)

相关推荐

  • Kotlin协程launch原理详解

    目录 正文 launch使用 launch原理 CoroutineStart中找invoke方法 startCoroutineCancellable逻辑 小结 正文 launch我们经常用,今天来看看它是什么原理. 建议: 食用本篇文章之前记得先食用Kotlin协程之createCoroutine和startCoroutine launch使用 launch我们应该很熟悉了,随便举个例子: fun main() { val coroutineScope = CoroutineScope(Job(

  • kotlin 协程上下文异常处理详解

    目录 引言 一.协程上下文 1.CoroutineContext 2.CorountineScope 3.子协程继承父协程 二.协程的异常传递 1.协程的异常传播 2.不同上下文(没有继承关系)之间协程异常会怎么样? 3.向用户暴露异常 三.协程的异常处理 使用SupervisorJob 异常捕获器CoroutineExceptionHandler Android中全局异常的处理 引言 从前面我们可以大致了解了协程的玩法,如果一个协程中使用子协程,那么该协程会等待子协程执行结束后才真正退出,而达

  • Kotlin 协程的取消机制详细解读

    目录 引言 协程的状态 取消协程的用法 协程取消的有效性 如何写出可以取消的代码 在 finally 中释放资源 使用不可取消的 block CancellationException 超时取消 异步的超时和资源 取消检查的底层原理 引言 在 Java 语言中提供了线程中断的能力,但并不是所有的线程都可以中断的,因为 interrupt 方法并不是真正的终止线程,而是将一个标志位标记为中断状态,当运行到下一次中断标志位检查时,才能触发终止线程. 但无论如何,终止线程是一个糟糕的方案,因为在线程的

  • Kotlin协程之Flow基础原理示例解析

    目录 引言 一.Flow的创建 二.Flow的消费 1.SafeFlow类 2.AbstractFlow类 3. SafeCollector类 4.消费过程中的挂起 引言 本文分析示例代码如下: launch(Dispatchers.Main) { flow { emit(1) emit(2) }.collect { delay(1000) withContext(Dispatchers.IO) { Log.d("liduo", "$it") } Log.d(&qu

  • Kotlin协程的基础与使用示例详解

    目录 一.协程概述 1.概念 2.特点 3.原理 1)续体传递 2)状态机 二.协程基础 1.协程的上下文 2.协程的作用域 3.协程调度器 4.协程的启动模式 5.协程的生命周期 1)协程状态的转换 2)状态标识的变化 三.协程使用 1.协程的启动 1)runBlocking方法 2)launch方法 3)async方法 4)suspend关键字 5)withContext方法 6)suspend方法 2.协程间通信 1)Channel 2)Channel的容量 3)produce方法与act

  • Kotlin协程launch启动流程原理详解

    目录 1.launch启动流程 反编译后的Java代码 2.协程是如何被启动的 1.launch启动流程 已知协程的启动方式之一是Globalscope.launch,那么Globalscope.launch的流程是怎样的呢,直接进入launch的源码开始看起. fun main() { coroutineTest() Thread.sleep(2000L) } val block = suspend { println("Hello") delay(1000L) println(&q

  • kotlin之协程的理解与使用详解

    前言         为什么在kotlin要使用协程呢,这好比去了重庆不吃火锅一样的道理.协程的概念并不陌生,在python也有提及.任何事务的作用大多是对于所依赖的环境相应而生的,协程对于kotlin这门语言也不例外.协程的优点,总的来说有如下几点:轻量级,占用更少的系统资源: 更高的执行效率: 挂起函数较于实现Runnable或Callable接口更加方便可控: kotlin.coroutine 核心库的支持,让编写异步代码更加简单.当然在一些不适应它的用法下以上优势也会成为劣势. 1.协程

  • Kotlin协程的启动方式介绍

    目录 1.GlobalScope.launch 2.runBlocking 启动协程 3.async启动协程 启动协程的基本方式 1.GlobalScope.launch 代码示例: fun testGlobalScope() { GlobalScope.launch { println("Coroutinue started!") delay(1000L) println("Hello World!") } println("After launch!&

  • 关于Python核心框架tornado的异步协程的2种方法详解

    什么是异步? 含义 :双方不需要共同的时钟,也就是接收方不知道发送方什么时候发送,所以在发送的信息中就要有提示接收方开始接收的信息,如开始位,同时在结束时有停止位 现象:没有共同的时钟,不考虑顺序来了就处理 直观感受:就是不用等了,效率高 同步 含义:指两个或两个以上随时间变化的量在变化过程中保持一定的相对关系 现象:有一个共同的时钟,按来的顺序一个一个处理 直观感受 :就是需要等候,效率低下 那么今天我们看怎么用2种方法用代码实现tornado的异步? 这些是导入的包: 2种方法用代码实现to

  • Android 应用程序的启动流程示例详解

    目录 应用进程的启动流程 1.ActivityStackSupervisor.startSpecificActivity 2.ATMS.startProcessAsync 3.LocalService.startProcess 4.startProcessLocked函数 5.ProcessList.startProcessLocked 6.ProcessList.startProcessLocked重载 7.ProcessList.startProcess 8.ZygoteState.star

  • Python协程的用法和例子详解

    从句法上看,协程与生成器类似,都是定义体中包含 yield 关键字的函数.可是,在协程中, yield 通常出现在表达式的右边(例如, datum = yield),可以产出值,也可以不产出 -- 如果 yield 关键字后面没有表达式,那么生成器产出 None. 协程可能会从调用方接收数据,不过调用方把数据提供给协程使用的是 .send(datum) 方法,而不是next(-) 函数. ==yield 关键字甚至还可以不接收或传出数据.不管数据如何流动, yield 都是一种流程控制工具,使用

  • Swoole4.4协程抢占式调度器详解

    前言 Swoole内核团队开设的专栏,会逐渐投入精力写文章介绍Swoole的开发历程,实现原理,应用实践等,大家可以更好的交流,共同学习,建设PHP生态. 协程调度 去年Swoole推出了4.0版本后,完整的支持PHP协程,我们可以基于协程实现CSP编程,身边的开发者惊呼,原来PHP代码还可以这样写.Swoole的协程默认是基于IO调度,程序中有阻塞会自动让出当前协程,协程的各种优势我们不在这里展开讨论.如果是IO密集型的场景,可以表现得很不错.但是对于CPU密集型的场景,会导致一些协程因为得不

  • Spring Boot面试必问之启动流程知识点详解

    目录 一 面试提问 1.1 Spring Boot启动流程 1.2 SpringBoot自动装配 二 知识点详解 2.1 SpringBoot核心注解: 2.2详细启动流程(结合源码) 总结 一 面试提问 1.1 Spring Boot启动流程 ???面试官:说说SpringBoot启动流程吧 ?? 我 : 首先从main找到run()方法,在执行run()方法之前new一个SpringApplication对象 进入run()方法,创建应用监听器SpringApplicationRunList

  • Kotlin协程Dispatchers原理示例详解

    目录 前置知识 demo startCoroutineCancellable intercepted()函数 DefaultScheduler中找dispatch函数 Runnable传入 Worker线程执行逻辑 小结 前置知识 Kotlin协程不是什么空中阁楼,Kotlin源代码会被编译成class字节码文件,最终会运行到虚拟机中.所以从本质上讲,Kotlin和Java是类似的,都是可以编译产生class的语言,但最终还是会受到虚拟机的限制,它们的代码最终会在虚拟机上的某个线程上被执行. 之

随机推荐