百度360必应搜狗淘宝本站头条
python判断字典是否为空crontab每周一执行aes和des区别安装指定版本npmadb pull
当前位置：网站首页 > IT知识 > 正文

直击灵魂一问:协程到底是怎么切换线程的?

liuian 2025-03-25 15:18 9 浏览

前置知识

CoroutineScope到底是什么?

CoroutineScope即协程运行的作用域,它的源码很简单

public interface CoroutineScope {
    public val coroutineContext: CoroutineContext
}

可以看出CoroutineScope的代码很简单，主要作用是提供CoroutineContext，协程运行的上下文 我们常见的实现有GlobalScope,LifecycleScope,ViewModelScope等

GlobalScope与ViewModelScope有什么区别?

public object GlobalScope : CoroutineScope {
    /**
     * 返回 [EmptyCoroutineContext].
     */
    override val coroutineContext: CoroutineContext
        get() = EmptyCoroutineContext
}

public val ViewModel.viewModelScope: CoroutineScope
    get() {
        val scope: CoroutineScope? = this.getTag(JOB_KEY)
        if (scope != null) {
            return scope
        }
        return setTagIfAbsent(
            JOB_KEY,
            CloseableCoroutineScope(SupervisorJob() + Dispatchers.Main.immediate)
        )
    }

两者的代码都挺简单，从上面可以看出

  1. GlobalScope返回的为CoroutineContext的空实现
  2. ViewModelScope则往CoroutineContext中添加了Job与Dispatcher

我们先来看一段简单的代码

 fun testOne(){
  GlobalScope.launch {
            print("1:" + Thread.currentThread().name)
            delay(1000)
            print("2:" + Thread.currentThread().name)
        }
 }
 //打印结果为：DefaultDispatcher-worker-1
    fun testTwo(){
        viewModelScope.launch {
            print("1:" + Thread.currentThread().name)
            delay(1000)
            print("2:" + Thread.currentThread().name)
        }
    }
    //打印结果为: main

上面两种Scope启动协程后，打印当前线程名是不同的，一个是线程池中的一个线程，一个则是主线程 这是因为ViewModelScope在CoroutineContext中添加了
Dispatchers.Main.immediate的原因

我们可以得出结论：协程就是通过Dispatchers调度器来控制线程切换的

什么是调度器？

从使用上来讲，调度器就是我们使用的Dispatchers.Main,Dispatchers.Default，Dispatcher.IO等 从作用上来讲，调度器的作用是控制协程运行的线程 从结构上来讲，Dispatchers的父类是ContinuationInterceptor,然后再继承于CoroutineContext 它们的类结构关系如下：

这也是为什么Dispatchers能加入到CoroutineContext中的原因,并且支持+操作符来完成增加

什么是拦截器

从命名上很容易看出，ContinuationInterceptor即协程拦截器，先看一下接口

interface ContinuationInterceptor : CoroutineContext.Element {
    // ContinuationInterceptor 在 CoroutineContext 中的 Key
    companion object Key : CoroutineContext.Key
    /**
     * 拦截 continuation
     */
    fun  interceptContinuation(continuation: Continuation): Continuation

    //...
}

从上面可以提炼出两个信息

  1. 拦截器的Key是单例的，因此当你添加多个拦截器时，生效的只会有一个
  2. 我们都知道，Continuation在调用其Continuation#resumeWith()方法，会执行其suspend修饰的函数的代码块，如果我们提前拦截到，是不是可以做点其他事情？这就是调度器切换线程的

原理

上面我们已经介绍了是通过Dispatchers指定协程运行的线程，通过interceptContinuation在协程恢复前进行拦截，从而切换线程

带着这些前置知识，我们一起来看下协程启动的具体流程，明确下协程切换线程源码具体实现

协程线程切换源码分析

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
}

总共有3个参数：

  1. 传入的协程上下文
  2. CoroutinStart启动器，是个枚举类，定义了不同的启动方法，默认是CoroutineStart.DEFAULT
  3. block就是我们传入的协程体，真正要执行的代码

这段代码主要做了两件事:

  1. 组合新的CoroutineContext
  2. 再创建一个 Continuation

组合新的CoroutineContext

public actual fun CoroutineScope.newCoroutineContext(context: CoroutineContext): CoroutineContext {
    val combined = coroutineContext + context
    val debug = if (DEBUG) combined + CoroutineId(COROUTINE_ID.incrementAndGet()) else combined
    return if (combined !== Dispatchers.Default && combined[ContinuationInterceptor] == null)
        debug + Dispatchers.Default else debug
}

从上面可以提炼出以下信息:

  1. 会将launch方法传入的context与CoroutineScope中的context组合起来
  2. 如果combined中没有拦截器，会传入一个默认的拦截器，即Dispatchers.Default,这也解释了为什么我们没有传入拦截器时会有一个默认切换线程的效果

创建一个Continuation

val coroutine = if (start.isLazy)
        LazyStandaloneCoroutine(newContext, block) else
        StandaloneCoroutine(newContext, active = true)
    coroutine.start(start, coroutine, block)

默认情况下，我们会创建一个StandloneCoroutine 值得注意的是，这个coroutine其实是我们协程体的complete，即成功后的回调，而不是协程体本身 然后调用coroutine.start，这表明协程开始启动了

协程的启动

public fun  start(start: CoroutineStart, receiver: R, block: suspend R.() -> T) {
    initParentJob()
    start(block, receiver, this)
}

接着调用CoroutineStart的start来启动协程，默认情况下调用的是CoroutineStart.Default

经过层层调用，最后到达了：

internal fun  (suspend (R) -> T).startCoroutineCancellable(receiver: R, completion: Continuation) =
    runSafely(completion) {
        // 外面再包一层 Coroutine
        createCoroutineUnintercepted(receiver, completion)
            // 如果需要，做拦截处理
            .intercepted()
            // 调用 resumeWith 方法      
            .resumeCancellableWith(Result.success(Unit))
    }

这里就是协程启动的核心代码，虽然比较短，却包括3个步骤：

  1. 创建协程体Continuation
  2. 创建拦截 Continuation,即DispatchedContinuation
  3. 执行DispatchedContinuation.resumeWith方法

创建协程体Continuation

调用
createCoroutineUnintercepted,会把我们的协程体即suspend block转换成Continuation,它是SuspendLambda，继承自ContinuationImpl
createCoroutineUnintercepted方法在源码中找不到具体实现，不过如果你把协程体代码反编译后就可以看到真正的实现

创建DispatchedContinuation

public actual fun  Continuation.intercepted(): Continuation =
    (this as? ContinuationImpl)?.intercepted() ?: this

//ContinuationImpl
public fun intercepted(): Continuation =
        intercepted
            ?: (context[ContinuationInterceptor]?.interceptContinuation(this) ?: this)
                .also { intercepted = it }     

//CoroutineDispatcher
public final override fun  interceptContinuation(continuation: Continuation): Continuation =
      DispatchedContinuation(this, continuation)

从上可以提炼出以下信息

  1. interepted是个扩展方法，最后会调用到ContinuationImpl.intercepted方法
  2. 在intercepted会利用CoroutineContext，获取当前的拦截器
  3. 因为当前的拦截器是CoroutineDispatcher，因此最终会返回一个DispatchedContinuation，我们其实也是利用它实现线程切换的
  4. 我们将协程体的Continuation传入DispatchedContinuation,这里其实用到了装饰器模式，实现功能的增强

这里其实很明显了，通过DispatchedContinuation装饰原有协程，在DispatchedContinuation里通过调度器处理线程切换，不影响原有逻辑，实现功能的增强

拦截处理

    //DispatchedContinuation
    inline fun resumeCancellableWith(
        result: Result,
        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)
                }
            }
        }
    }

上面说到了启动时会调用DispatchedContinuation的resumeCancellableWith方法 这里面做的事也很简单：

  1. 如果需要切换线程，调用dispatcher.dispatcher方法，这里的dispatcher是通过CoroutineConext取出来的
  2. 如果不需要切换线程，直接运行原有线程即可

调度器的具体实现

我们首先明确下，CoroutineDispatcher是通过CoroutineContext取出来的,这也是协程上下文作用的体现 CoroutineDispater官方提供了四种实现：Dispatchers.Main,Dispatchers.IO,Dispatchers.Default,Dispatchers.Unconfined 我们一起简单看下Dispatchers.Main的实现

internal class HandlerContext private constructor(
    private val handler: Handler,
    private val name: String?,
    private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
    public constructor(
        handler: Handler,
        name: String? = null
    ) : this(handler, name, false)

    //...

    override fun dispatch(context: CoroutineContext, block: Runnable) {
        // 利用主线程的 Handler 执行任务
        handler.post(block)
    }
}

可以看到，其实就是用handler切换到了主线程 如果用Dispatcers.IO也是一样的，只不过换成线程池切换了

如上所示，其实就是一个装饰模式

  1. 调用CoroutinDispatcher.dispatch方法切换线程
  2. 切换完成后调用DispatchedTask.run方法，执行真正的协程体

delay是怎样切换线程的?

上面我们介绍了协程线程调度的基本原理与实现,下面我们来回答几个小问题 我们知道delay函数会挂起，然后等待一段时间再恢复。可以想象，这里面应该也涉及到线程的切换，具体是怎么实现的呢？

public suspend fun delay(timeMillis: Long) {
    if (timeMillis <= 0) return // don't delay
    return suspendCancellableCoroutine sc@ { cont: CancellableContinuation ->
        // if timeMillis == Long.MAX_VALUE then just wait forever like awaitCancellation, don't schedule.
        if (timeMillis < Long.MAX_VALUE) {
            cont.context.delay.scheduleResumeAfterDelay(timeMillis, cont)
        }
    }
}

internal val CoroutineContext.delay: Delay get() = get(ContinuationInterceptor) as? Delay ?: DefaultDelay

Dealy的代码也很简单，从上面可以提炼出以下信息 delay的切换也是通过拦截器来实现的，内置的拦截器同时也实现了Delay接口 我们来看一个具体实现

internal class HandlerContext private constructor(
    private val handler: Handler,
    private val name: String?,
    private val invokeImmediately: Boolean
) : HandlerDispatcher(), Delay {
    override fun scheduleResumeAfterDelay(timeMillis: Long, continuation: CancellableContinuation) {
        // 利用主线程的 Handler 延迟执行任务，将完成的 continuation 放在任务中执行
        val block = Runnable {
            with(continuation) { resumeUndispatched(Unit) }
        }
        handler.postDelayed(block, timeMillis.coerceAtMost(MAX_DELAY))
        continuation.invokeOnCancellation { handler.removeCallbacks(block) }
    }

    //..
}
  1. 可以看出，其实也是通过handler.postDelayed实现延时效果的
  2. 时间到了之后，再通过resumeUndispatched方法恢复协程
  3. 如果我们用的是Dispatcher.IO，效果也是一样的，不同的就是延时效果是通过切换线程实现的

withContext是怎样切换线程的?

我们在协程体内，可能通过withContext方法简单便捷的切换线程，用同步的方式写异步代码，这也是kotin协程的主要优势之一

    fun test(){
        viewModelScope.launch(Dispatchers.Main) {
            print("1:" + Thread.currentThread().name)
            withContext(Dispatchers.IO){
                delay(1000)
                print("2:" + Thread.currentThread().name)
            }
            print("3:" + Thread.currentThread().name)
        }
    }
    //1，2，3处分别输出main,DefaultDispatcher-worker-1,main

可以看出这段代码做了一个切换线程然后再切换回来的操作，我们可以提出两个问题

  1. withContext是怎样切换线程的?
  2. withContext内的协程体结束后，线程怎样切换回到Dispatchers.Main?
public suspend fun  withContext(
    context: CoroutineContext,
    block: suspend CoroutineScope.() -> T
): T {  
    return suspendCoroutineUninterceptedOrReturn sc@ { uCont ->
        // 创建新的context
        val oldContext = uCont.context
        val newContext = oldContext + context
        ....
        //使用新的Dispatcher，覆盖外层
        val coroutine = DispatchedCoroutine(newContext, uCont)
        coroutine.initParentJob()
        //DispatchedCoroutine作为了complete传入
        block.startCoroutineCancellable(coroutine, coroutine)
        coroutine.getResult()
    }
}

private class DispatchedCoroutine(
    context: CoroutineContext,
    uCont: Continuation
) : ScopeCoroutine(context, uCont) {
 //在complete时会会回调
    override fun afterCompletion(state: Any?) {
        afterResume(state)
    }

    override fun afterResume(state: Any?) {
        //uCont就是父协程，context仍是老版context,因此可以切换回原来的线程上
        uCont.intercepted().resumeCancellableWith(recoverResult(state, uCont))
    }
}

这段代码其实也很简单，可以提炼出以下信息

  1. withContext其实就是一层Api封装，最后调用到了startCoroutineCancellable,这就跟launch后面的流程一样了，我们就不继续跟了
  2. 传入的context会覆盖外层的拦截器并生成一个newContext，因此可以实现线程的切换
  3. DispatchedCoroutine作为complete传入协程体的创建函数中，因此协程体执行完成后会回调到afterCompletion中
  4. DispatchedCoroutine中传入的uCont是父协程，它的拦截器仍是外层的拦截器，因此会切换回原来的线程中

总结

本文主要回答了kotlin协程到底是怎么切换线程的这个问题，并对源码进行了分析 简单来讲主要包括以下步骤：

  1. 向CoroutineContext添加Dispatcher，指定运行的协程
  2. 在启动时将suspend block创建成Continuation,并调用intercepted生成DispatchedContinuation
  3. DispatchedContinuation就是对原有协程的装饰，在这里调用Dispatcher完成线程切换任务后，resume被装饰的协程，就会执行协程体内的代码了

其实kotlin协程就是用装饰器模式实现线程切换的 看起来似乎有不少代码，但是真正的思路其实还是挺简单的，这大概就是设计模式的作用吧

原文链接：
https://juejin.cn/post/6981056016897015838

相关推荐

GANs为何引爆机器学习?这篇基于TensorFlow的实例教程为你解惑!

「机器人圈导览」：生成对抗网络无疑是机器学习领域近三年来最火爆的研究领域，相关论文层出不求，各种领域的应用层出不穷。那么，GAN到底如何实践？本文编译自Medium，该文作者以一朵玫瑰花为例，详细阐...

高丽大学等机构联合发布StarGAN:可自定义表情和面部特征

原文来源：arXiv、GitHub作者：YunjeyChoi、MinjeChoi、MunyoungKim、Jung-WooHa、SungKim、JaegulChoo「雷克世界」编译：嗯~...

TensorFlow和PyTorch相继发布最新版，有何变化

原文来源：GitHub「机器人圈」编译：嗯~阿童木呀、多啦A亮Tensorflow主要特征和改进在Tensorflow库中添加封装评估量。所添加的评估量列表如下：1.深度神经网络分类器（DNNCl...

「2022 年」崔庆才 Python3 爬虫教程 - 深度学习识别滑动验证码缺口

上一节我们使用OpenCV识别了图形验证码躯壳欧。这时候就有朋友可能会说了，现在深度学习不是对图像识别很准吗？那深度学习可以用在识别滑动验证码缺口位置吗？当然也是可以的，本节我们就来了解下使用深度...

20K star!搞定 LLM 微调的开源利器

LLM（大语言模型）微调一直都是老大难问题，不仅因为微调需要大量的计算资源，而且微调的方法也很多，要去尝试每种方法的效果，需要安装大量的第三方库和依赖，甚至要接入一些框架，可能在还没开始微调就已经因为...

大模型DeepSeek本地部署后如何进行自定义调整?

1.理解模型架构a)查看深度求索官方文档或提供的源代码文件，了解模型的结构、输入输出格式以及支持的功能。模型是否为预训练权重？如果是，可以在预训练的基础上进行微调（Fine-tuning）。是否需要...

因配置不当，约5000个AI模型与数据集在公网暴露

除了可访问机器学习模型外，暴露的数据还可能包括训练数据集、超参数，甚至是用于构建模型的原始数据。前情回顾·人工智能安全动态向ChatGPT植入恶意“长期记忆”，持续窃取用户输入数据多模态大语言模型的致...

基于pytorch的深度学习人员重识别

基于pytorch的深度学习人员重识别Torchreid是一个库。基于pytorch的深度学习人员重识别。特点：支持多GPU训练支持图像的人员重识别与视频的人员重识别端到端的训练与评估简单的re...

DeepSeek本地部署:轻松训练你的AI模型

引言：为什么选择本地部署？在AI技术飞速发展的今天，越来越多的企业和个人希望将AI技术应用于实际场景中。然而，对于一些对数据隐私和计算资源有特殊需求的用户来说，云端部署可能并不是最佳选择。此时，本地部...

谷歌今天又开源了，这次是Sketch-RNN

前不久，谷歌公布了一项最新技术，可以教机器画画。今天，谷歌开源了代码。在我们研究其代码之前，首先先按要求设置Magenta环境。（https://github.com/tensorflow/magen...

Tensorflow 使用预训练模型训练的完整流程

前面已经介绍了深度学习框架Tensorflow的图像的标注和训练数据的准备工作，本文介绍一下使用预训练模型完成训练并导出训练的模型。1.选择预训练模型1.1下载预训练模型首先需要在Tensorf...

30天大模型调优学习计划（30分钟训练大模型）

30天大模型调优学习计划，结合Unsloth和Lora进行大模型微调，掌握大模型基础知识和调优方法，熟练应用。第1周：基础入门目标：了解大模型基础并熟悉Unsloth等工具的基本使用。Day1：大模...

python爬取喜马拉雅音频，json参数解析

一.抓包分析json，获取加密方式1.抓包获取音频界面f12打开抓包工具，播放一个（非vip）视频，点击“媒体”单击打开可以复制URL，发现就是我们要的音频。复制“CKwRIJEEXn-cABa0Tg...

五、JSONPath使用(Python)（json数据python）

1.安装方法pipinstalljsonpath2.jsonpath与Xpath下面表格是jsonpath语法与Xpath的完整概述和比较。Xpathjsonpath概述/$根节点.@当前节点...

Python网络爬虫的时候json=就是让你少写个json.dumps()

大家好，我是皮皮。一、前言前几天在Python白银交流群【空翼】问了一个Python网络爬虫的问题，提问截图如下：登录请求地址是这个：二、实现过程这里【甯同学】给了一个提示，如下所示：估计很多小伙伴和...

一周热门
最近发表
标签列表