【问题标题】:How does Kotlin coroutines know when to yield when making network calls?Kotlin 协程在进行网络调用时如何知道何时让步?
【发布时间】:2019-02-23 02:48:22
【问题描述】:

我是 Kotlin 协程的新手,但我没有弄清楚的一件事是,协程在进行网络调用时如何知道何时让步给其他人。

如果我理解正确的话,协程是抢占式工作的,这意味着它知道当它需要执行一些耗时的任务(通常是 I/O 操作)时,何时让步给其他协程。

例如,假设我们想要绘制一些 UI 来显示来自远程服务器的数据,并且我们只有一个线程来调度我们的协程。我们可以启动一个协程进行 REST API 调用以获取数据,同时让另一个协程绘制不依赖数据的 UI 的其余部分。但是,由于我们只有一个线程,因此一次只能运行一个协程。除非用于获取数据的协程在等待数据到达时抢先让步,否则这两个协程将按顺序执行。

据我所知,Kotlin 的协程实现并没有修补任何现有的 JVM 实现或 JDK 网络库。因此,如果协程正在调用 REST API,它应该像使用 Java 线程一样阻塞。我这么说是因为我在 python 中似乎有类似的概念,它们被称为绿色线程。为了让它与 python 的内置网络库一起工作,必须首先对网络库进行“猴子补丁”。对我来说,这是有道理的,因为只有网络库本身知道何时让步。

那么谁能解释一下 Kotlin 协程在调用阻塞 Java 网络 API 时如何知道何时让步?或者如果没有,那么是否意味着上面示例中提到的任务不能同时执行给单个线程?

谢谢!

【问题讨论】:

  • Kotlin 使用非阻塞 io 进行网络操作。也没有人阻止库根据需要创建尽可能多的线程。 Wiki it:非阻塞 I/O (Java)

标签: java kotlin coroutine kotlin-coroutines


【解决方案1】:

协程抢先工作

不。使用协程,您只能实现协作多线程,在这种情况下,您可以通过显式方法调用来暂停和恢复协程。协程只关注按需挂起和恢复,而协程调度器负责确保它在适当的线程上启动和恢复。

研究这段代码将帮助你了解 Kotlin 协程的精髓:

import kotlinx.coroutines.experimental.*
import kotlin.coroutines.experimental.*

fun main(args: Array<String>) {
    var continuation: Continuation<Unit>? = null
    println("main(): launch")
    GlobalScope.launch(Dispatchers.Unconfined) {
        println("Coroutine: started")
        suspendCoroutine<Unit> {
            println("Coroutine: suspended")
            continuation = it
        }
        println("Coroutine: resumed")
    }
    println("main(): resume continuation")
    continuation!!.resume(Unit)
    println("main(): back after resume")
}

这里我们使用最简单的Unconfined 调度程序,它不进行任何调度,它在你调用launch { ... }continuation.resume() 的地方运行协程。协程通过调用suspendCoroutine 暂停自身。这个函数通过传递你以后可以用来恢复协程的对象来运行你提供的块。我们的代码将其保存到var continuation。控制权返回到launch 之后的代码,我们使用延续对象来恢复协程。

整个程序在主线程上执行并打印:

main(): launch
Coroutine: started
Coroutine: suspended
main(): resume continuation
Coroutine: resumed
main(): back after resume

我们可以启动一个协程来进行 REST API 调用以获取数据,同时让另一个协程绘制 UI 的其余部分,而不依赖于数据。

这实际上描述了您将使用普通线程做什么。协程的优点是您可以在 GUI 绑定代码的中间进行“阻塞”调用,它不会冻结 GUI。在您的示例中,您将编写一个进行网络调用然后更新 GUI 的协程。当网络请求正在进行时,协程被挂起,其他事件处理程序运行,使 GUI 保持活动状态。处理程序不是协程,它们只是常规的 GUI 回调。

用最简单的话,你可以写出这样的 Android 代码:

activity.launch(Dispatchers.Main) {
    textView.text = requestStringFromNetwork()
}

...

suspend fun requestStringFromNetwork() = suspendCancellableCoroutine<String> {
    ...
}

requestStringFromNetwork 相当于“修补 IO 层”,但您实际上并没有修补任何东西,您只是围绕 IO 库的公共 API 编写包装器。几乎所有 Kotlin IO 库都添加了这些包装器,并且还有 Java IO 库的扩展库。如果您关注these instructions,您也可以自己编写。

【讨论】:

  • 感谢您的澄清。你的回答让我重新思考了我的问题。我还发现此链接很有帮助link
  • 所以这是我现在的理解,1. 使用 Kotlin 协程不会自动将阻塞调用转换为非阻塞调用,它只是提供了一种更简单、更自然的方式来编写异步代码(通过提供一种机制用于挂起和恢复协程)。 2. 为了充分利用它,我们最好使用某种异步库,例如Java NIO,或具有回调的CompletableFuture。 3. 另一方面,如果使用阻塞调用,除非我们使用更多线程,否则我们仍然会被阻塞。 @Marko,您能否验证我的新理解是否正确?谢谢!
  • 是的,几乎达到了目标。阻塞调用总是以native 方法结束,你无法从外部修补它们以将它们变成非阻塞调用。 Python 在这里很不一样,因为它的 IO 一直都是基于非阻塞 IO 和 IO 事件循环。这是你可以修补以获得类似协程的行为。
  • 但是,我不建议您直接接触 Java NIO,它是非常低级且非常尴尬的 API。你应该使用像 Netty 这样的库,让它变得更方便。
  • 谢谢。我再次阅读了您的第一个答案,现在我对它的理解要好得多。其实我相信它首先回答了我的问题,只是我当时没有得到它。
【解决方案2】:

答案是:协程不知道网络调用或 I/O 操作。你必须根据自己的需要编写代码,将繁重的工作封装到不同的协程中,以便它们可以并发执行,因为默认行为是顺序的。

例如:

suspend fun doSomethingUsefulOne(): Int {
    delay(1000L) // pretend we are doing something useful here (maybe I/O)
    return 13
}

suspend fun doSomethingUsefulTwo(): Int {
    delay(1000L) // pretend we are doing something useful here (maybe I/O), too
    return 29
}

fun main(args: Array<String>) = runBlocking<Unit> {
        val time = measureTimeMillis {
            val one = doSomethingUsefulOne()
            val two = doSomethingUsefulTwo()
            println("The answer is ${one + two}")
        }
    println("Completed in $time ms")
}

会产生这样的结果:

The answer is 42
Completed in 2017 ms

doSomethingUsefulOne() 和 doSomethingUsefulTwo() 将按顺序执行。 如果你想要并发执行,你必须改写:

fun main(args: Array<String>) = runBlocking<Unit> {
    val time = measureTimeMillis {
        val one = async { doSomethingUsefulOne() }
        val two = async { doSomethingUsefulTwo() }
        println("The answer is ${one.await() + two.await()}")
    }
    println("Completed in $time ms")
}

这将产生:

The answer is 42
Completed in 1017 ms

因为 doSomethingUsefulOne() 和 doSomethingUsefulTwo() 将同时执行。

来源:https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#composing-suspending-functions

更新: 关于协程的执行位置,我们可以在 github 项目指南https://github.com/Kotlin/kotlinx.coroutines/blob/master/coroutines-guide.md#thread-local-data 中阅读:

有时传递一些线程本地数据的能力很方便,但是对于没有绑定到任何特定线程的协程,如果不编写大量样板文件,就很难手动实现。

对于 ThreadLocal 来说,asContextElement 扩展函数就是来救命的。它创建了一个额外的上下文元素,它保留给定 ThreadLocal 的值,并在每次协程切换其上下文时恢复它。

很容易在实际中演示它:

val threadLocal = ThreadLocal<String?>() // declare thread-local variable
fun main(args: Array<String>) = runBlocking<Unit> {
    threadLocal.set("main")
    println("Pre-main, current thread: ${Thread.currentThread()}, threadlocal value: '${threadLocal.get()}'")
    val job = launch(Dispatchers.Default + threadLocal.asContextElement(value = "launch")) {
        println("Launch start, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
        yield()
        println("After yield, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
    }
    job.join()
    println("Post-main, current thread: ${Thread.currentThread()}, thread local value: '${threadLocal.get()}'")
}

在这个例子中,我们使用 Dispatchers.Default 在后台线程池中启动新的协程,因此它可以在与线程池不同的线程上工作,但它仍然具有我们使用 threadLocal 指定的线程局部变量的值.asContextElement(value = "launch"),不管协程在哪个线程上执行。因此,输出(带调试)是:

Pre-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'
Launch start, current thread: Thread[CommonPool-worker-1 @coroutine#2,5,main], thread local value: 'launch'
After yield, current thread: Thread[CommonPool-worker-2 @coroutine#2,5,main], thread local value: 'launch'
Post-main, current thread: Thread[main @coroutine#1,5,main], thread local value: 'main'

【讨论】:

  • 您好雷蒙德,感谢您的回复。您能否详细说明您上面给出的示例之间的差异。我检查了文档,在我看来,“等待”调用意味着等待协程完成而不阻塞当前线程(我没有找到任何“bg”通过的文档)。但是,正如我在问题中所说的那样,如果两个协程不知道何时让步,而不引入多个线程,它们如何可能同时运行?
  • 特别是在您的示例中,即使“downloadBigFileUsingNetwork”作为协程显式执行,我希望它继续使用当前线程,除非它显式屈服于负责渲染的父协程用户界面。理想情况下,它应该在等待下载完成时让步。但是正如你提到的,它不知道它正在执行 I/O 操作,这意味着它不会自动产生。在这种情况下,我想它仍然会阻塞当前线程,不是吗?
  • 您的问题可以简化为:downloadBigFileUsingNetwork 是否对非阻塞网络调用使用阻塞?所有的 Android 网络库都是异步的,所以你只需要使用它们的回调来恢复协程。如果您仍然使用阻塞 API,那么您必须使用 withContext(Default) { blockingCall() },它将阻塞调用变成暂停调用。
  • 您的回答中没有任何内容表明协程的暂停。您的代码具有完全相同的形式和行为,就好像所有调用都被阻塞一样。你可以用val future = threadPool.submit { task() }task.await()future.get() 来实现你的async { task() }
  • @MarkoTopolnik 抱歉,我可能误解了这个问题的重点,跨语言障碍很难克服。
猜你喜欢
  • 1970-01-01
  • 2019-09-09
  • 2023-03-17
  • 2020-02-23
  • 2015-08-31
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
  • 1970-01-01
相关资源
最近更新 更多