使用协程并行执行作业

Question

我想使用协程并行执行多个作业。这是我想出的一段代码。
我有 2 个疑问：

• 如何确保在调用者线程中发生完成回调？

• 代码变得更像我以前使用的回调模式 正常线程。请建议更改设计以实现 协程可读性优势。

class ParallelExecutor {

    suspend fun <OUTPUT> execute(
        jobs: List<suspend () -> OUTPUT>,
        onTimeout: (jobIndex: Int) -> OUTPUT,
        onFailure: (jobIndex: Int, exception: Throwable) -> OUTPUT,
        onCompletion: suspend (jobIndex: Int, result: OUTPUT) -> Unit,
        timeout: Long,
        onFullCompletion: suspend () -> Unit = {},
        invokeDispatcher: CoroutineDispatcher = Dispatchers.Default
    ) {
        withContext(invokeDispatcher) {
            var counter = 0
            val listenJobs = mutableListOf<Deferred<OUTPUT>>()

            jobs.forEachIndexed { index, job ->
                val listenJob = async {

                    try {
                        job()
                    } catch (e: Exception) {
                        onFailure(index, e)
                    }
                }
                listenJobs.add(listenJob)
            }

            listenJobs.forEachIndexed { index, job ->
                launch {
                    val output = try {
                        withTimeout(timeout) {
                            job.await()
                        }
                    } catch (e: TimeoutCancellationException) {
                        onTimeout(index)
                    }
                    onCompletion(index, output)
                    if (++counter == listenJobs.size) {
                        onFullCompletion()
                    }
                }
            }
        }
    }
}

Answer 1

在我看来，您可以大大简化您的代码。您不需要先启动所有async 作业然后启动更多作业以等待它们的两步习惯用法。您可以只 launch 作业并委托给同一块内的回调。这样，回调自然会在调用者的调度程序上被调用，并且只有作业本身可以在更改后的上下文中使用invokeDispatcher 调用。

onFullCompletion 看起来像一段属于调用方的代码，位于execute 调用下方。由于execute 不会抛出任何异常，因此您不需要任何try-finally 来获取它。

suspend fun <OUTPUT> execute(
    jobs: List<suspend () -> OUTPUT>,
    onTimeout: (jobIndex: Int) -> OUTPUT,
    onFailure: (jobIndex: Int, exception: Throwable) -> OUTPUT,
    onCompletion: suspend (jobIndex: Int, result: OUTPUT) -> Unit,
    timeout: Long,
    invokeDispatcher: CoroutineDispatcher = Dispatchers.Default
) {
    coroutineScope {
        jobs.mapIndexed { index, job ->
            launch {
                val output = try {
                    withTimeout(timeout) {
                        withContext(invokeDispatcher) {
                            job()
                        }
                    }
                } catch (e: TimeoutCancellationException) {
                    onTimeout(index)
                } catch (e: Exception) {
                    onFailure(index, e)
                }
                onCompletion(index, output)
            }
        }
    }
}

Answer 2

进行了一些改进以回答您的疑问。

class ParallelExecutor {

    suspend fun <OUTPUT> execute(
        jobs: List<suspend () -> OUTPUT>,
        onTimeout: (jobIndex: Int) -> OUTPUT,
        onFailure: (jobIndex: Int, exception: Throwable) -> OUTPUT,
        onCompletion: suspend (jobIndex: Int, result: OUTPUT) -> Unit,
        timeout: Long,
        invokeDispatcher: CoroutineDispatcher = Dispatchers.Default
    ) {
        supervisorScope {
            val listenJobs = jobs.map { job ->
                async(invokeDispatcher) {
                    withTimeout(timeout) {
                        job()
                    }
                }
            }

            listenJobs.forEachIndexed { index, job ->
                launch {
                    val output = try {
                        job.await()
                    } catch (e: TimeoutCancellationException) {
                        onTimeout(index)
                    } catch (e: Exception) {
                        onFailure(index, e)
                    }
                    onCompletion(index, output)
                }
            }
        }
    }
}

• 现在在超时时取消作业。
• 现在在调用者的 dispatcher 中调用完成回调。
• 已修复决定何时调用 onFullCompletion 时的竞争条件。
• 删除了一些您并不真正需要的方法。

如果您觉得这更像是回调模式，那么您根本不应该使用回调。协程的设计使您可以在使用站点以最少的样板编写此类代码，因此这样的函数不是必需的，而且看起来很奇怪（恕我直言）。