Android应用中如何使用Kotlin协程实现类似RxJava Observable.amb的多数据源并行取最快结果功能？

嘿，这个需求我之前也碰到过，用Kotlin协程实现其实挺顺手的，正好可以用select表达式配合协程作用域来搞定，和RxJava的amb()效果完全一致，我给你一步步拆解下～

1. 先把数据源调整成带返回值的形式 #

首先得把你的挂起函数改成能返回实际数据的样子（我这里假设返回String，你可以根据业务需求调整类型）：

suspend fun dataSourceOne(): String {
    delay(1_000L)
    return "Result from Data Source 1"
}

suspend fun dataSourceTwo(): String {
    delay(2_000L)
    return "Result from Data Source 2"
}

suspend fun dataSourceThree(): String {
    delay(3_000L)
    return "Result from Data Source 3"
}

2. 核心实现逻辑：并行调用+取最快结果+自动取消 #

这里的关键是coroutineScope和select的组合，代码写出来很简洁：

suspend fun getFastestResult(): String = coroutineScope {
    // 启动三个并行的异步任务
    val task1 = async { dataSourceOne() }
    val task2 = async { dataSourceTwo() }
    val task3 = async { dataSourceThree() }

    // 用select监听第一个完成的任务，拿到结果就返回
    select<String> {
        task1.onAwait { it }
        task2.onAwait { it }
        task3.onAwait { it }
    }
}

3. 验证取消是否生效（可选） #

如果你想确认其他调用真的被取消了，可以给数据源函数加个取消监听：

suspend fun dataSourceTwo(): String {
    try {
        delay(2_000L)
        return "Result from Data Source 2"
    } catch (e: CancellationException) {
        println("Data Source 2 was cancelled")
        throw e // 必须重新抛出取消异常，不然协程会被误判为正常完成
    }
}

suspend fun dataSourceThree(): String {
    try {
        delay(3_000L)
        return "Result from Data Source 3"
    } catch (e: CancellationException) {
        println("Data Source 3 was cancelled")
        throw e
    }
}

调用getFastestResult()后，控制台会打印出另外两个数据源被取消的日志，说明逻辑生效了。

为啥这个方案好使？ #

• coroutineScope就像协程的大管家，所有在它里面启动的子协程都归它管。当我们通过select拿到第一个结果返回时，coroutineScope会立刻结束，并且自动取消所有还在运行的子协程，完全不用手动处理取消逻辑。
• select表达式专门用来监听多个挂起操作，哪个先完成就取哪个的结果，完美匹配你要的「最快返回」需求。

Android场景下的注意事项 #

• 一定要在合适的协程作用域里调用这个方法：比如在ViewModel里用viewModelScope，在Activity/Fragment里用lifecycleScope，这样能自动和组件生命周期绑定，避免内存泄漏。
• 如果你的数据源是网络请求这类IO操作，记得给async指定Dispatchers.IO调度器，比如：

  val task1 = async(Dispatchers.IO) { dataSourceOne() }
  

举个ViewModel里的使用例子：

class MyViewModel : ViewModel() {
    fun fetchFastestData() {
        viewModelScope.launch {
            val fastestResult = getFastestResult()
            // 拿到结果后更新UI或者做其他业务处理
        }
    }

    private suspend fun getFastestResult(): String = coroutineScope {
        val task1 = async(Dispatchers.IO) { dataSourceOne() }
        val task2 = async(Dispatchers.IO) { dataSourceTwo() }
        val task3 = async(Dispatchers.IO) { dataSourceThree() }

        select<String> {
            task1.onAwait { it }
            task2.onAwait { it }
            task3.onAwait { it }
        }
    }
}

这样就完全实现了你要的功能，和RxJava的amb()效果一致，还更贴合Kotlin协程的风格～

内容的提问来源于stack exchange，提问作者Maxim Firsoff