主页 / coding / 问题 / 79556930
Accepted
Ken Kiarie
Asked: 2025-04-05 20:07:36 +0800 CST

如何生成新的协同程序来处理冷流中的每个元素？

  • 772

遇到了一个例子，模拟访问一个慢速数据库以获取与可通过更慢的网络资源访问的配置文件相关联的用户标识符流：

fun getAllUserIds():Flow<Int>{
    return flow{
        repeat(3){
            delay(200.milliseconds)
            log("Emitting!")
            emit(it)
            }
         } 
      }

suspend fun getProfileFromNetwork(id:Int):String{
    delay(2.seconds)
    return "Profile[$id]"
  }

fun main(){
    val ids = getAllUserIds()
    runBlocking{
        ids
            .map{getProfileFromNetwork(it)}
            .collect{log("Got $it")}
        }
  }

ID 的发布和配置文件的请求是相互交织的。我该如何通过为流程中的每个元素的处理生成新的协程来优化这一点？

kotlin-coroutines

2 个回答

  1. Best Answer

    如果您想要获得整个结果，则可以使用flatMapMerge，它是为这种精确场景设计的：同时将流中的每个元素映射到新流并合并结果。

    fun getAllUserIds(): Flow<Int> {
    return flow {
        repeat(3) {
            delay(200.milliseconds)
            log("Emitting!")
            emit(it)
        }
    }
}

suspend fun getProfileFromNetwork(id: Int): String {
    delay(2.seconds)
    return "Profile[$id]"
}

fun main() = runBlocking {
    getAllUserIds()
        .flatMapMerge { id ->
            flow {
                emit(getProfileFromNetwork(id))
            }
        }
        .collect { log("Got $it") }
}

    但我想到了一个更好的方法；向用户显示尽可能多的数据，并具有视觉效果，指示某些数据块尚未加载，例如使用ShimmerEffect或任何其他替代方法。尝试为列表的每个项目延迟加载它。

    当从 RecyclerView/List 项逻辑内的数据库获取 UserID 时，尝试触发一个独立的协程来获取 Profile（不要忘记采用缓存策略来最小化网络负载）。

    • 2

  2. 我认为您混淆了数据库流返回的内容。如果您查询所有用户 ID，您将得到一个List<Int>。当支持流程时，您可能还可以选择获取一个。每当数据库中发生某些变化Flow<List<Int>>时，流程都会发出一个新值。发生这种情况时，查询将自动再次执行，并返回所有用户 ID 的新列表。流程只是一种方便的方式来观察数据库随时间的变化。它不会一次发出每个用户 ID。

    考虑到这一点，你的虚拟函数getAllUserIds()实际上必须看起来像这样：

    fun getAllUserIds(): Flow<List<Int>> {
    return flow {
        repeat(3) {
            delay(200.milliseconds)
            log("Emitting!")
            val ids = (0..it).toList()
            emit(ids)
        }
    }
}

    此代码模拟每 ​​200 毫秒将一个新用户输入数据库。该流程现在将发出三个不同的用户列表：

    [0]
[0, 1]
[0, 1, 2]

    然后，您的主函数必须迭代列表来检索每个 id 的配置文件：

    fun main() {
    val flowOfIds = getAllUserIds()
    runBlocking {
        flowOfIds
            .map { listOfIds ->
                listOfIds.map { getProfileFromNetwork(it) }
            }
            .collect { log("Got $it") }
    }
}

    flowOfIds.map替换每个流值（即，如果是 a ，则流现在包含 aList<String>而不是List<Int>）。 listOfIds.map替换每个列表条目（即，列表现在包含Strings 而不是Ints ）。

    回到最初的问题。

    就像现在一样，getProfileFromNetwork()是按顺序调用的。如果您希望同时发生这种情况，则需要为每个操作启动一个新的协程：

    flowOfIds
    .map { listOfIds ->
        listOfIds
            .map { async { getProfileFromNetwork(it) } }
            .awaitAll()
    }

    async在 提供的范围内启动一个新的协程runBlocking，并awaitAll()等待所有协程完成。它返回一个配置文件列表，即List<String>。

    对每个新的流值重复此操作，因此0总共检索用户 ID 的配置文件三次，用户 ID 的配置文件2仅检索一次，因为它仅出现在最新的流发射中。

    由于getProfileFromNetwork比发出新流值的速度慢，因此您可能需要从 切换flowOfIds.map到flowOfIds.mapLatest。当新的流值到达时，这会取消之前的流转换，因此您不必等待之前的转换完成后再开始新的转换。

    此外，您应该认真考虑缓存 的结果getProfileFromNetwork()，这样您就不会重复发出相同的网络请求（例如0三次获取用户 ID 的个人资料）。这应该在 内部完成getProfileFromNetwork()。您可能应该将该函数移动到存储库类或类似的东西，以便您可以在那里构建缓存。

    • 1

相关问题