Kotlin延迟的anyOf等价

Question

协程async返回Deferred<T>，并且有延迟执行的示例和await的用法。

但是，我们如何等待任何一个Deffered实例完成呢？

简而言之

  // whats the equivalent of CompletableFuture.anyOf(...)?
  // is this how we do it? if so how costly is this?
  select<Unit> {
     deffered1.onAwait {}
     deffered2.onAwait {}
  }

Answer 1

可能不是最安全的方法，但这样的方法应该可以工作：

inline suspend fun <T> Iterable<Deferred<T>>.awaitAny(): T {
    var completed: T? = null
    forEachIndexed { index, deferred ->
        deferred.invokeOnCompletion {
            completed = deferred.getCompleted()
            forEachIndexed { index2, deferred2 ->
                if (index != index2) {
                    deferred2.cancel(it)
                }
            }
        }
    }
    forEach {
        try {
            it.await()
        } catch (ignored: JobCancellationException) {
            // ignore
        }
    }
    return completed!!
}

打样:下面打印1000张

launch(CommonPool) {
        // 10 - 1 second(s)
        val deferredInts = List(10, {
            val delayMs = (10 - it) * 1000
            async(CommonPool) {
                delay(delayMs)
                delayMs
            }
        })
        val first = deferredInts.awaitAny()
        println(first)
    }

Answer 2

使用如下select表达式

        val deferred1: Deferred<String?> = GlobalScope.async { getValue1() }
        val deferred2: Deferred<String?> = GlobalScope.async { getValue2() }
        val deferred3: Deferred<String?> = GlobalScope.async { getValue3() }

        val deferredList = listOf(deferred1, deferred2, deferred3)
        val firstCompletedResult = select<String?> {
            deferredList.forEach {
                it.onAwait {}
            }
        }
        
        Log.d(TAG, "firstCompleted: $firstCompletedResult")

Answer 3

对于多样性，提到select()的.merge(): Flow替代方案，我用它来处理a similar problem (获取延迟集合中的第一个，以非空值完成)，即

val firstCompletedResult = deferreds.map { it::await.asFlow() }.merge().first()

或者如果你喜欢的话

suspend fun <T> Iterable<Deferred<T>>.awaitAny(): T = map { it::await.asFlow() }.merge().first()