收集多个异步结果导致 Kotlin 协程忽略超时异常

Question

我有一个 Generator 类，它基本上会生成一些数据，例如：

interface Generator {
    suspend fun generate(): String?
}

有多种实现方式。其中一些可能会抛出异常，而其中一些可能需要很长时间才能生成数据：

class Faulty : Generator {
    override suspend fun generate(): String? {
        println("Faulty")
        throw IllegalArgumentException();
    }
}

class Lingering : Generator {
    override suspend fun generate(): String? {
        println("Lingering")
        delay(Duration.ofHours(1))
        return null
    }
}

但有些实现是值得的

class Good : Generator {
    override suspend fun generate(): String {
        println("Good")
        return "Goooood"
    }
}

我需要做的是收集由预配置生成器列表生成的数据，为每个生成器设置一个超时 generate 并忽略异常（但记录它们）：

fun main() = runBlocking {
    val generators = listOf(Faulty(), Lingering(), Good())
    val results = supervisorScope {
        generators
                .map { generator ->
                    async(CoroutineExceptionHandler { context, exception ->
                        println(exception)
                    }) {
                        withTimeoutOrNull(5000) {
                            generator.generate()
                        }
                    }
                }
                .awaitAll()
                .filterNotNull()
    }

    println(results)
}

问题是这段代码fails有异常：

Faulty
Lingering
Good
Exception in thread "main" java.lang.IllegalArgumentException
 at Faulty.generate (File.kt:12) 
 at FileKt$main$1$results$1$1$2$1.invokeSuspend (File.kt:41) 
 at FileKt$main$1$results$1$1$2$1.invoke (File.kt:-1) 

为什么supervisorScope 没有捕捉到它？我做错了什么？

Answer 1

来自 CoroutineExceptionHandler 的文档：

协程上下文中的可选元素，用于处理未捕获异常。

和

使用async 创建的协程始终捕获其所有异常并将它们表示在生成的 Deferred 对象中，因此它不会导致未捕获的异常。

因此，您的 async 作业不会发出未捕获的异常。稍后发生的awaitAll() 调用会再次引发异常。您已将未捕获的异常处理程序仅放在您的 async 上下文中，因此不会使用它。

此外，子协程无论如何都不会发出未捕获的异常。 Their exceptions are delegated up to their root ancestor.

正如 here 在上一节题为受监督协程中的异常 中所解释的那样，监督作用域的子级必须有一个使用该处理程序的根协程。

您可以做的是将整个任务包装在使用处理程序的launch 块中。由于某种原因，在runBlocking 上安装处理程序不起作用。也许这不算是根工作？

fun main() = runBlocking{
    val job = GlobalScope.launch(CoroutineExceptionHandler { context, exception ->
        println(exception)
    }) {
        val generators = listOf(Faulty(), Lingering(), Good())
        val results =
            supervisorScope {
                generators
                    .map { generator ->
                        async {
                            withTimeoutOrNull(5000) {
                                generator.generate()
                            }
                        }
                    }
                    .awaitAll()
                    .filterNotNull()
            }

        println(results)
    }
    job.join()
}

但我认为您引入 CoroutineExceptionHandler 的唯一原因可能是为了忽略异常。该策略不起作用，因为处理程序只处理未捕获的异常，这意味着恢复为时已晚。那时这项工作已经失败了。您必须将 generate() 调用包装在 try/catch 或 runCatching 中的 async 块中。