启动一些协程，并在超时的情况下将它们全部加入(不取消) [英] Launch a number of coroutines and join them all with timeout (without cancelling)

问题描述

我需要启动许多作业，这些作业将返回结果.

I need to launch a number of jobs which will return a result.

在主代码(不是 协程)中，启动工作后，我需要等待所有工作完成任务 OR 给定的超时时间到期，以先到者为准.

In the main code (which is not a coroutine), after launching the jobs I need to wait for them all to complete their task OR for a given timeout to expire, whichever comes first.

如果我因为所有作业都在超时之前完成而退出等待，那太好了，我将收集他们的结果.

If I exit from the wait because all the jobs completed before the timeout, that's great, I will collect their results.

但是，如果某些作业花费的时间超过超时时间，则我的主要功能需要在超时到期后立即唤醒，检查哪些作业及时完成(如果有)，哪些仍在运行，然后从在那里，没有取消仍在运行的作业.

But if some of the jobs are taking longer that the timeout, my main function needs to wake as soon as the timeout expires, inspect which jobs did finish in time (if any) and which ones are still running, and work from there, without cancelling the jobs that are still running.

您将如何编码这种等待?

How would you code this kind of wait?

推荐答案

解决方案直接来自问题.首先，我们将为任务设计一个挂起函数.让我们看看我们的要求:

The solution follows directly from the question. First, we'll design a suspending function for the task. Let's see our requirements:

如果某些作业花费的时间超过了超时时间...而没有取消仍在运行的作业.

if some of the jobs are taking longer that the timeout... without cancelling the jobs that are still running.

这意味着我们启动的作业必须是独立的(而不是子级)，因此我们将选择退出结构化并发，并使用GlobalScope启动它们，手动收集所有作业.我们使用async协程生成器是因为我们计划稍后收集其某些R类型的结果:

It means that the jobs we launch have to be standalone (not children), so we'll opt-out of structured concurrency and use GlobalScope to launch them, manually collecting all the jobs. We use async coroutine builder because we plan to collect their results of some type R later:

val jobs: List<Deferred<R>> = List(numberOfJobs) { 
    GlobalScope.async { /* our code that produces R */ }
}

启动作业后，我需要等待它们全部完成任务或等待指定的超时时间，以先到者为准.

after launching the jobs I need to wait for them all to complete their task OR for a given timeout to expire, whichever comes first.

让我们等待所有这些，并等待超时:

Let's wait for all of them and do this waiting with timeout:

withTimeoutOrNull(timeoutMillis) { jobs.joinAll() }

如果其中一项作业失败，我们将使用joinAll(而不是awaitAll)来避免异常，并使用withTimeoutOrNull来避免超时时出现异常.

We use joinAll (as opposed to awaitAll) to avoid exception if one of the jobs fail and withTimeoutOrNull to avoid exception on timeout.

我的主要功能需要在超时到期后立即唤醒，检查哪些作业已及时完成(如果有)以及哪些作业仍在运行

my main function needs to wake as soon as the timeout expires, inspect which jobs did finish in time (if any) and which ones are still running

jobs.map { deferred -> /* ... inspect results */ }

在主代码中(不是协程)...

In the main code (which is not a coroutine) ...

由于我们的主代码不是协程，因此必须以阻塞的方式等待，因此我们将使用runBlocking编写的代码桥接在一起.全部放在一起:

Since our main code is not a coroutine it has to wait in a blocking way, so we bridge the code we wrote using runBlocking. Putting it all together:

fun awaitResultsWithTimeoutBlocking(
    timeoutMillis: Long,
    numberOfJobs: Int
) = runBlocking {
    val jobs: List<Deferred<R>> = List(numberOfJobs) { 
        GlobalScope.async { /* our code that produces R */ }
    }    
    withTimeoutOrNull(timeoutMillis) { jobs.joinAll() }
    jobs.map { deferred -> /* ... inspect results */ }
}

P.S.我不建议在任何严重的生产环境中部署这种解决方案，因为在超时后让后台作业运行(泄漏)总是会在以后给您带来严重的伤害.仅在您完全了解这种方法的所有缺陷和风险时，才这样做.

P.S. I would not recommend deploying this kind of solution in any kind of a serious production environment, since letting your background jobs running (leak) after timeout will invariably badly bite you later on. Do so only if you throughly understand all the deficiencies and risks of such an approach.

这篇关于启动一些协程，并在超时的情况下将它们全部加入(不取消)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！