Kotlin Coroutines 1.5: GlobalScope 标记为 delicate，完善的 Channels API 等等

Lei

Read this post in other languages:

• English
,
• Deutsch
,
• Français
,
• 日本語
,
• 한국어
,
• Русский

合著者: Svetlana Isakova

Kotlin Coroutines 1.5.0 已发布! 以下为新版本带来的新特性：

• GlobalScope API已被标记为delicate。GlobalScope作为高级的API很容易被滥用。 在可能会被滥用的地方，现在编译器将发出警告，并要求您在程序中选择性引入该类。
• JUnit扩展。CoroutinesTimeout已可在JUnit5中使用。
• 完善的Channel API。以及针对库函数新的命名方案，引入了非挂起函数trySend和tryReceive以作为offer和poll更好的替代。
• 稳定的Reactive Integrations。我们添加了更多用于将Reactive Streams类型转换成Kotlin Flow的函数，很多现有的函数和ReactiveContext API已变得稳定。

开始使用Coroutines 1.5.0

GlobalScope API 已被标记为 delicate

现在 GlobalScope 类已被 @DelicateCoroutinesApi 注解所标记。 从现在开始，所有使用 GlobalScope 的地方都需要 @OptIn(DelicateCoroutinesApi::class) 显式引入。

虽然大多数情况下都不建议使用 GlobalScope，但官方文档仍通过这个精细的 API 引入这些概念。

全局 CoroutineScope 不会和任何job绑定。 GlobalScope 多用于启动运行在整个应用生命周期内且不会提前取消的顶级协程。 在 GlobalScope 中启动的活动中协程不会让该进程保持存活状态。 就像守护线程一样。

源于 GlobalScope 的精密，其使用容易导致资源或内存泄露。 从 GlobalScope 中启动的协程并不遵守结构化并发的原则，一旦因为某种原因(例如较慢的网速)被挂起或延迟，它会持续运行并消耗资源。 例如以下代码：

fun loadConfiguration() {
    GlobalScope.launch {
        val config = fetchConfigFromServer() // network request
        updateConfiguration(config)
    }
}

调用 loadConfiguration 会在 GlobalScope 中创建一个后台工作的协程，并且不会等待其取消或完成。 一旦网络很慢，它会一直在后台等待并且消耗资源。 重复代用 loadConfiguration 将消耗更多的资源。

可行的替代

在许多情况下，应避免使用 GlobalScope，并且其包含的函数应标记为 suspend，例如：

suspend fun loadConfiguration() {
    val config = fetchConfigFromServer() // network request
    updateConfiguration(config)
}

如果通过 GlobalScope.launch 启动多个并发操作，则应将相关的操作通过 coroutineScope 进行分组：

// concurrently load configuration and data
suspend fun loadConfigurationAndData() {
    coroutineScope {
        launch { loadConfiguration() }
        launch { loadData() }
    }
}

从顶层代码的非挂起上下文中启动并发操作时，应该使用有所限制的 CoroutineScope 实例来代替 GlobalScope。

合理的用法

仅仅有限的场景下，可以合理且安全地使用 GlobalScope，例如必须在整个应用生命周期过程中保持活动的顶层后台进程。 因此，所有使用 GlobalScope 的地方都需要 @OptIn(DelicateCoroutinesApi::class) 显式引入，例如：

// A global coroutine to log statistics every second, 
// must be always active
@OptIn(DelicateCoroutinesApi::class)
val globalScopeReporter = GlobalScope.launch {
    while (true) {
        delay(1000)
        logStatistics()
    }
}

我们建议您仔细审查所有对 GlobalScope 的用法，并注解那些属于“合理用例”的用法。 对于其他用法，它们很可能是代码中的 bug —— 如上所述来替换掉 GlobalScope 的用法。

JUnit5 扩展

我们添加了允许在独立线程中运行测试的 CoroutinesTimeout 注解，并在限制的时间之后让测试失效并中断线程。 在之前，CoroutinesTimeout 只在 JUnit4 中可用。 在这个正式版中，我们已将其集成到 JUnit5 中了。

要使用这个新的注解，请在您的项目中添加下依赖：

dependencies {
  …
  testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-debug:$coroutinesVersion")
}

这是一个简单的示例，说明如何在测试中使用新的 CoroutinesTimeout：

import kotlinx.coroutines.debug.junit5.CoroutinesTimeout
import kotlinx.coroutines.delay
import kotlinx.coroutines.runBlocking
import org.junit.Test

@CoroutinesTimeout(100)
class CoroutinesTimeoutSimpleTest {

     @CoroutinesTimeout(300)
     @Test
     fun firstTest() {
         runBlocking {
             delay(200)  // succeeds
         }
     }

     @Test
     fun secondTest() {
         runBlocking {
             delay(200)  // fails
         }
     }
 }

在该示例中，CoroutinesTimeout是类级别注解，并且还标记了 firstTest方法。 被注解的测试方法不会超时，因为函数级别的注解覆盖了类级别的注解。 而secondTest用到了类级别的注解，因此它会超时。

注解以下述形式定义：

package kotlinx.coroutines.debug.junit5

public annotation class CoroutinesTimeout(
    val testTimeoutMs: Long,
    val cancelOnTimeout: Boolean = false
)

第一个参数testTimeoutMs，以毫秒级定义了超时时间。 第二个参数cancelOnTimeout，定义了是否在超时后取消所有正在运行的协程。 如果设置为true，则所有协程将被自动取消。

每当您使用CoroutinesTimeout注解，它会自动启用协程调试器并在超时时转储所有协程。 其转储包含了协程创建的堆栈轨迹。 如果有需要禁用对创建的堆栈跟踪以加快测试速度，请直接使用CoroutinesTimeoutExtension，它允许进行配置。

非常感谢 Abhijit Sarkar，他为 JUnit 5 的 CoroutinesTimeout 构建了一个实用的的 PoC。 这个想法被开发成为我们在1.5版本中添加的 CoroutinesTimeout 新注解。

完善的 Channel API

通道是重要的通信原语，可让您可以在不同的协程和回调之间传递数据。 在这个版本中，我们重新设计了部分 Channel API，用更好的选择来替换了引起混淆的 offer 和 poll 函数。 在这个过程中，我们为挂起和非挂起方法设计了一种新的统一命名方案。

新的命名方案

我们试图建立统一的命名方案，以便后续在其他库或Coroutines API中使用。 我们想要确保函数的名称能够传达其行为的信息。 结果是，我们提出了如下内容：

• 常规的挂起方法保持不变，例如，send，receive。
• 封装了异常的相应非挂起方法始终以”try”为前缀：trySend 和 tryReceive，而非旧的 offer 和 poll。
• 新的封装异常的挂起方法将以”Catching”作为后缀。

让我们来深入了解这些新方法的细枝末节。

Try 函数：send 和 receive 的非挂起版本

一个协程可以向通道发送一些信息，而另一个协程可以从该通道接收这个信息。 send 和 receive 函数都是挂起的。 如果通道已满并且不能接受新的元素，则 send 挂起其协程，而如果通道中没有返回元素，则 receive 挂起其协程：

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        // Suspends until the element can be sent
        println("Sending...")
        channel.send("Element")
     }
     // Suspends until the element can be received
     println("Receiving...")
     println(channel.receive())
}

这些函数拥有可在同步代码中使用的非挂起形式：offer和poll，但已被废弃，现在推荐使用trySend和tryReceive函数。 让我们讨论这种变动的原因。

offer 和 poll 与 send 和 receive 做了相同的事，但不会引发挂起。 这听起来很简单，而且在元素可被收发时一切正常。 一旦抛出异常，会发生什么呢？ send 和 receive 会挂起直到它们能继续任务。 一旦通道已满无法添加元素，或通道为空无法检索任何元素，offer 和 poll 会简单地返回 false 和 null。 它们都会在关闭了的通道里尝试运行时抛出异常，但最后结果却令人迷惑。

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        println("Sending...")
        // Doesn't suspend
        // Returns 'false' if the channel is full
        // Or throws an exception if it's closed
        channel.offer("Element")
    }
    println("Receiving...")
    // Doesn't suspend
    // Returns 'null' if the channel is empty
    println(channel.poll())
//  channel.close()
}

在这个示例中，在元素添加前调用 poll 将会立即返回 null。 请注意不该这样使用：您应该持续定期地轮询元素，我们是为了简化说明才直接调用。 offer 的调用也是不成功的，因为我们的通道是一个 rendezvous 类型的通道，并且其缓冲区容量为零。 结果 offer 返回了 false，而 poll 返回了 null，仅仅是因为它们以错误的顺序被调用。

在上述示例中，取消 channel.close()语句的注释，将会抛出异常。 在这个例子中，poll 仍然会像之前一样返回 false。 但当 offer 尝试向已关闭的通道中增加一个元素时，则会抛出异常。 我们收到了很多抱怨，表示这种行为容易出错。 这很容易便忘记去捕获该异常，又您宁可忽略抑或以其他方式去处理，但它会令您的程序崩溃。

新的trySend和tryReceive修复了这个问题，并返回了包含更多细节的结果。 每个返回的ChannelResult实例，为以下三种其中之一：成功的结果，失败或通道已关闭的标记。

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    launch {
        println("Sending...")
        // Doesn't suspend
        // Returns 'Failed' if the channel is full
        // Or 'Channel was closed' result if it's closed
        val result = channel.trySend("Element")
        println(result)

        //We can verify the result
        if(result.isClosed){
            println("Sending failed. The channel is closed.")
        }
    }
    println("Receiving...")
    println(channel.tryReceive())
//  channel.close()
}

该示例的工作方式与上一个相同，区别在于 tryReceive 和 trySend 返回了更详尽的结果。 您可以在输出中看到 Value(Failed)，而不是 false 和 null。 再次取消关闭通道语句的注释，现在 trySend 会返回 Closed 异常被捕获的结果。

多亏了内联值类，ChannelResult 不会有额外的封装类，并且如果返回的是成功值，则原样返回，而不会产生任何开销。

异常捕获函数：带异常封装的挂起函数

从这个版本开始，带异常封装的挂起方法将以”Catching”作为后缀。 例如，新的 receiveCatching 函数会处理通道关闭情况下的异常。 请思考这个简单的例子：

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    channel.close()
    println(channel.receiveCatching())
}

在我们尝试接收值之前通道已被关闭。 但是程序成功执行完毕，表明该通道已关闭。 如果将 receiveCatching 换成普通的 receive 函数，它将抛出 ClosedReceiveChannelException：

import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.launch
import kotlinx.coroutines.runBlocking

fun main() = runBlocking<Unit> {
    val channel = Channel<String>()
    channel.close()
    println(channel.receive())
}

目前我们只提供了 receiveCatching 和 onReceiveCatching(而非之前内部的 receiveOrClosed )，但我们打算添加更多函数。

迁移您的代码至新函数

您可以用新的快捷执行自动替换项目中 offer 和 poll 函数的所有用法。 由于 offer 返回了布尔值，因此其等效替换为 channel.trySend(“ Element”).isSuccess。

同样，poll 函数返回可空的元素，因此将其替换为 channel.tryReceive().getOrNull()。

如果返回的结果并没有被使用，则可以将其直接替换为新的调用。

现在对于异常的处理有所不同，因此也许需要您手动进行必要的更新。 如果您的代码依赖于 ‘offer’ 和 ‘poll’ 方法在已关闭通道上抛出的异常，则需要使用以下替代方法。

channel.offer("Element") 的等效替换在已关闭的通道会抛出异常，即便通道是正常关闭：

channel
  .trySend("Element")
  .onClosed { throw it ?: ClosedSendChannelException("Channel was closed") }
  .isSuccess

如果通道由于错误而关闭，则 channel.poll() 的等效替换将引抛出异常，如果正常关闭，则返回 null：

channel.tryReceive()
  .onClosed { if (it != null) throw it }
  .getOrNull()

这样的变动反映了 offer 和 poll 函数旧的行为。

我们假定在大多数情况下，您的代码不会依赖于已关闭通道的细微之处，而是依赖它本身是错误的根源。 因此，IDE 提供的自动替换功能简化了语义。 如果这并不符合您的情况，请手动审查并更新，并考虑完全进行重写，以不同方式处理通道已被关闭的情况，而不是抛出异常。

响应式集成迈向稳定之旅

对于负责响应式框架集成的大部分函数，Kotlin Coroutines 的 1.5 版本将其升级到稳定 API。

在JVM生态系统，只有少数框架可以处理符合Reactive Streams标准的异步流。 例如，Project Reactor和RxJava是该领域中最流行的两个Java框架。

而Kotlin Flow有所不同，并且其类型与标准指定的不兼容，但从概念上来讲它们仍然是流。 可将Flow转换为响应式(规范且兼容TCK的)Publisher，反之亦然。 这些开箱即用的转换器是由kotlinx.coroutines提供的，可以在相应的响应式模块中找到。

例如，如果您需要与Project Reactor类型的互操作性，则应在项目中添加以以下依赖：

dependencies {          
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:${coroutinesVersion}")
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:${coroutinesVersion}")
}

然后，如果需要 Reactive Streams 类型，则使用 Flow<T>.asPublisher()，或 Flow<T>.asFlux()（如果您需要 Project Reactor 类型）。

// acquire a Flow instance
val flow: Flow<event> = flow { … }

// Convert Flow to Publisher
val publisher = flow.asPublisher()

// Convert Flow to Reactor's Flux
val flux = flow.asFlux()

//Convert back to Flow 
val anotherFlow = flux.asFlow()

这是该主题的一个概览。 如果您想了解更多信息，请考虑阅读Roman Elizarov’s的Reactive Streams and Kotlin Flows的文章。

虽然与响应式库的集成正朝着API稳定的方向努力，但从技术上讲，目标是摆脱 @ExperimentalCoroutinesApi并实现各主题的剩余内容。

改善与 Reactive Streams 的集成

为了确保第三方框架与 Kotlin Coroutines 之间的互操作性，与 Reactive Streams 规范的兼容性很重要。 这对于在老旧项目中无需重写任何代码便能应采用 Kotlin Coroutines 很有帮助。

这次我们设法将大量的函数提升到稳定版。 现在可以将任何 Reactive Streams 类型转换为 Flow 并返回。 例如新代码可以用 Coroutines 编写，但可以通过反向转换器与旧的响应式代码库进行集成：

fun legacyFunThatHaveToReturnObservable(): Observable<int> {
  return flow<int> {
    // Use the power of flow!
  }
  // various flow operations
  .asObservable()
}

此外，对 ReactorContext 进行了大量的改进，将 Reactor 的 Context 包装到 CoroutineContext 中，以实现 Project Reactor 与 Kotlin Coroutines 的无缝集成。 通过这个集成，便能使用协程传递 Reactor 的上下文信息。

所有响应式集成的上下文都是通过订阅者的上下文隐式传播的，例如 Mono，Flux，Publisher.asFlow，Flow.asPublisher 和 Flow.asFlux。 这是一个将订阅者的 Context 传播到 ReactorContext 的简易示例：

import kotlinx.coroutines.currentCoroutineContext
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.reactor.ReactorContext
import kotlinx.coroutines.reactor.asFlux

fun main() {
    val flow = flow<int> {
       println("Reactor context in Flow: " +
          currentCoroutineContext()[ReactorContext]?.context)
    }

    // No context
    // prints "Reactor context in Flow: null"
    flow.asFlux().subscribe() 

    // Add subscriber's context
    // prints "Reactor context in Flow: Context1{answer=42}"
    flow.asFlux()
        .contextWrite { ctx -> ctx.put("answer", 42) }
        .subscribe() 
}

在上面的示例中，我们构造了一个 Flow 实例，然后将其转换为无上下文的 Reactor 的 Flux 实例。 调用不带参数的 subscribe()方法，其效果是获取发布者发送的所有数据。 结果是程序将打印出“Reactor context in Flow: null”。

下面的链式调用将 Flow 转换为 Flux，随后链式调用将一个键值对 answer = 42 添加到了 Reactor 的上下文中。 对 subscribe() 的调用触发了整个调用链。 在这种情况下，由于上下文已添加了数据，因此程序将打印”Reactor context in Flow: Context1{answer=42}”

新的快捷函数

在 Coroutines 上下文中使用诸如 Mono 之类的响应式类型时，有一些方便的函数可以在不阻塞线程的情况下进行检索。 在这个版本中，我们弃用了所有 Publisher 的 awaitSingleOr* 函数，和专门为 Mono 和 Maybe 设计的 await* 函数。

Mono 最多产生一个值，因此最后一个元素与第一个元素相同。 在这种情况下，删除剩余元素的语义是没有用的。 因此 Mono.awaitFirst() 和 Mono.awaitLast() 被弃用，取而代之的是 Mono.awaitSingle()。

开始使用 kotlinx.coroutines 1.5!

新版本有着令人印象深刻的变动列表。 在完善 Channels API 的同时开发新的命名方案是团队的一项显著成就。 同时我们非常注重让 Coroutines API 尽可能简单直观。

要开始使用 Kotlin Coroutines 的新版本，只需更新 build.gradle.kts 文件的内容。 首先，请确保您拥有最新版本的 Kotlin Gradle 插件：

plugins {
   kotlin("jvm") version "1.5.0"
}

然后更新依赖的版本，包括 Reactive Streams 特定集成的库。

val coroutinesVersion = "1.5.0"

dependencies { 
  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm:$coroutinesVersion")
  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactive:$coroutinesVersion")
  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-reactor:$coroutinesVersion")
  implementation("org.jetbrains.kotlinx:kotlinx-coroutines-rx3:$coroutinesVersion")
  ...
}

更多的观看及阅读材料

• Kotlin Coroutines 1.5.0 视频
• Coroutines 指南
• API 文档
• Kotlin Coroutines’ GitHub repository
• Coroutines 1.4.0 博文
• Kotlin 1.5.0 发布博文

如果您遇到任何问题

• GitHub问题跟踪器报告问题。
• 在Kotlin Slack上的#coroutines频道中寻求帮助(获得邀请)。

Subscribe to Kotlin Blog updates

Subscribe form

By submitting this form, I agree to the JetBrains Privacy Policy Notification icon

提交此表单，即表示我同意 JetBrains s.r.o. ("JetBrains") 使用我的姓名、电子邮件地址和位置数据向我发送简报和商业通讯，并为此目的而处理我的个人数据。我同意 JetBrains 根据 JetBrains 隐私政策为此目的使用第三方服务处理上述数据。我了解我可以在我的个人资料中随时撤回此同意。此外，每封电子邮件中也都包含退订链接。

Submit

谢谢！