kotlin - 如何正确加入 CoroutineScope 中启动的所有作业

Question

我正在重构一些目前在 GlobalScope 上启动协同程序的 Kotlin 代码到结构化的基于并发的方法。我需要在 JVM 退出之前加入在我的代码中启动的所有作业。我的类可以分解为以下界面:

interface AsyncTasker {
    fun spawnJob(arg: Long)
    suspend fun joinAll()
}

用法:

fun main(args: Array<String>) {
    val asyncTasker = createAsyncTasker()

    asyncTasker.spawnJob(100)
    asyncTasker.spawnJob(200)
    asyncTasker.spawnJob(300)
    asyncTasker.spawnJob(500)

    // join all jobs as they'd be killed when the JVM exits
    runBlocking {
        asyncTasker.joinAll()
    }
}

我的 GlobalScope基于的实现如下所示:

class GlobalScopeAsyncTasker : AsyncTasker {
    private val pendingJobs = mutableSetOf<Job>()

    override fun spawnJob(arg: Long) {
        var job: Job? = null
        job = GlobalScope.launch(Dispatchers.IO) {
            someSuspendFun(arg)
            pendingJobs.remove(job)
        }
        pendingJobs.add(job)
    }

    override suspend fun joinAll() {
        // iterate over a copy of the set as the
        // jobs remove themselves from the set when we join them
        pendingJobs.toSet().joinAll()
    }
}

显然，这并不理想，因为跟踪每个待处理的工作并不是很优雅，而且是旧的基于线程的编码范例的残余。
作为更好的方法，我正在创建自己的 CoroutineScope用于启动所有子项，提供 SupervisorJob .

class StructuredConcurrencyAsyncTasker : AsyncTasker {

    private val parentJob = SupervisorJob()
    private val scope = CoroutineScope(Dispatchers.IO + parentJob)

    override fun spawnJob(arg: Long) {
        scope.launch {
            someSuspendFun(arg)
        }
    }

    override suspend fun joinAll() {
        parentJob.complete() // <-- why is this needed??
        parentJob.join()
    }
}

在最初开发此解决方案时，我省略了对 parentJob.complete() 的调用。 , 导致 join()至 无限期暂停 .这感觉很不直观，所以我正在寻找确认/输入这是否是解决此类问题的正确方法。为什么我必须手动 complete() parent 的工作？有没有更简单的方法来解决这个问题？
Kotlin playground with the code

Answer 1

我想知道这种行为将来是否会改变。目前，链接问题中的答案仍然成立。暂时parentJob.join()不加入它的 child 。对我来说 Job#join() -documentation的以下部分是深入挖掘的原因:

Note that the job becomes complete only when all its children are complete.

请注意，启动的协程作业可能处于除 completed 之外的另一种状态。 .您可能想通过类似 parentJob.children.forEach { println(it) } 的方式来验证这一点。 (或任何您想检查或您可能想调试它的信息;-))在您的 parentJob.join() 之前-陈述。
有(至少？)两种方法可以确保所有启动的子协程作业都完成，这样它就不会在那时挂起或过早完成:

等待所有 child 作业首先完成(如评论中的链接答案中所述)，即:

parentJob.children.forEach { it.join() }

这不需要额外的 parentJob.join()或 parentJob.complete()因此可能是首选？ parentJob将在其所有子项完成时完成。

调用 complete调用前 join ， IE。:

parentJob.complete()
parentJob.join()

注意拨打 complete这里只是将状态转换为完成，正如 Job documentation 中所述.在完成状态下，它也将等待其子节点完成。如果您只是拨打 complete()没有 join该程序可能会在运行您启动的协程作业之前退出。如果你只是 join()正如您已经经历过的那样，它可能会无限期暂停。