用 Kotlin 协程实现 Observale.amb？答案

【问题标题】：Implement Observale.amb with Kotlin-coroutines?用 Kotlin 协程实现 Observale.amb？
【发布时间】：2021-10-09 03:38:08
【问题描述】：

例如，我有 3 个来自不同端点的数据源。我想并行调用所有这些并获得第一个答案（最快），然后应该丢弃其他调用。我知道如何使用带有Observable.amb() 的 RxJava。如何使用 Kotlin 协程实现它？重要的是 - 在第一个结果之后无需等待其他调用。

suspend fun dataSourceOne(){
   delay(1_000L)
}

suspend fun dataSourceTwo(){
   delay(2_000L)
}

suspend fun dataSourceThree(){
   delay(3_000L)
}

// should call [dataSourceOne(), dataSourceTwo(), dataSourceThree()] in parallel 
// and discard [dataSourceTwo(), dataSourceThree()] after the getting a result from dataSourceOne()

PS：Android 应用程序。

【问题讨论】：

标签： android rx-java2 kotlin-coroutines

【解决方案1】：

您可以启动收集器，使用原子整数来索引获胜者并跟踪所有其他要取消的作业。例如：

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.FlowCollector
import kotlinx.coroutines.flow.collect
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicInteger

@FlowPreview
class FlowAmbIterable<T>(private val sources: Iterable<Flow<T>>) : Flow<T> {
    @InternalCoroutinesApi
    override suspend fun collect(collector: FlowCollector<T>) {
        val winner = AtomicInteger()
        val jobs = ConcurrentHashMap<Job, Int>()
        coroutineScope {
            var i = 1
            for (source in sources) {
                val idx = i
                val job = launch {
                    source.collect {
                        val w = winner.get()
                        if (w == idx) {
                            collector.emit(it)
                        } else if (w == 0 && winner.compareAndSet(0, idx)) {
                            for (j in jobs.entries) {
                                if (j.value != idx) {
                                    j.key.cancel()
                                }
                            }

                            collector.emit(it)
                        } else {
                            throw CancellationException()
                        }
                    }
                }

                jobs[job] = i
                val w = winner.get()
                if (w != 0 && w != i) {
                    job.cancel()
                    break
                }

                i++
            }
        }
    }
}

【讨论】：

谢谢！也许有更优雅的解决方案？
如果有一个优雅的解决方案，kotlinx.coroutines 已经有这样的操作符了。

【解决方案2】：

参考文献

import kotlinx.coroutines.ExperimentalCoroutinesApi
import kotlinx.coroutines.channels.ChannelResult
import kotlinx.coroutines.channels.onFailure
import kotlinx.coroutines.channels.onSuccess
import kotlinx.coroutines.channels.produce
import kotlinx.coroutines.coroutineScope
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.emitAll
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.selects.select
import kotlinx.coroutines.yield

@ExperimentalCoroutinesApi
public fun <T> race(flows: Iterable<Flow<T>>): Flow<T> = flow {
  coroutineScope {
    // 1. Collect to all source Flows
    val channels = flows.map { flow ->
      // Produce the values using the default (rendezvous) channel
      produce {
        flow.collect {
          send(it)
          yield() // Emulate fairness, giving each flow chance to emit
        }
      }
    }

    // If channels List is empty, just return and complete result Flow.
    if (channels.isEmpty()) {
      return@coroutineScope
    }
    
    // If channels List has single element, just forward all events from it.
    channels
      .singleOrNull()
      ?.let { return@coroutineScope emitAll(it) }

    // 2. When a new event arrives from a source Flow, pass it down to a collector.
    // Select expression makes it possible to await multiple suspending functions simultaneously
    // and select the first one that becomes available.
    val (winnerIndex, winnerResult) = select<Pair<Int, ChannelResult<T>>> {
      channels.forEachIndexed { index, channel ->
        channel.onReceiveCatching {
          index to it
        }
      }
    }

    // 3. Cancel all other Flows.
    channels.forEachIndexed { index, channel ->
      if (index != winnerIndex) {
        channel.cancel()
      }
    }

    // 4. Forward all events from the winner Flow .
    winnerResult
      .onSuccess {
        emit(it)
        emitAll(channels[winnerIndex])
      }
      .onFailure {
        it?.let { throw it }
      }
  }
}

【讨论】：