Início / coding / Perguntas / 79556930
Accepted
Ken Kiarie
Asked: 2025-04-05 20:07:36 +0800 CST

Como gerar novas corrotinas para o processamento de cada elemento em um fluxo frio?

  • 772

Me deparei com um exemplo que simula o acesso a um banco de dados lento para obter um fluxo de identificadores de usuários associados a um perfil que pode ser acessado por meio de um recurso de rede ainda mais lento:

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")}
        }
  }

a emissão dos IDs e a solicitação de perfis estão interligadas. Como eu otimizaria isso gerando novas coroutines para o processamento de cada elemento no fluxo?

kotlin-coroutines

2 respostas

  1. Best Answer

    Se quiser obter o resultado como um todo, você pode usar flatMapMerge, que foi projetado para esse cenário exato: mapear simultaneamente cada elemento no fluxo para um novo fluxo e mesclar os resultados.

    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") }
}

    Mas eu penso em uma abordagem melhor; mostrar o máximo de dados buscados para o usuário e ter um efeito visual, indicando que algum pedaço de dados ainda está para carregar usando, por exemplo, ShimmerEffect ou qualquer outra alternativa. Tente carregá-lo preguiçosamente para cada item da lista.

    Tente disparar uma corrotina independente para buscar o Perfil quando o UserID tiver sido buscado do banco de dados dentro da lógica do item RecyclerView/List (não se esqueça de ter uma política de cache para minimizar a carga da rede).

    • 2

  2. Acho que você confundiu o que o fluxo do banco de dados retorna. Se você consultar todos os IDs de usuário, você obterá um List<Int>. Quando os fluxos são suportados, então você também pode ter a opção de obter um Flow<List<Int>>. O fluxo emitirá um novo valor sempre que algo mudar no banco de dados. Quando isso acontece, a consulta é executada automaticamente novamente e uma nova lista de todos os IDs de usuário é retornada. O fluxo é apenas uma maneira conveniente de observar o banco de dados ao longo do tempo para alterações. Ele não emite cada ID de usuário um de cada vez.

    Com isso em mente, sua função fictícia getAllUserIds()deve ser parecida com esta:

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

    Este código emula que a cada 200 ms um novo usuário é inserido no banco de dados. O fluxo agora emitirá três listas diferentes de usuários:

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

    Sua função principal deve então iterar a lista para recuperar o perfil de cada id:

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

    flowOfIds.mapsubstitui cada valor de fluxo (ou seja, o fluxo agora contém um List<String>em vez de um List<Int>). listOfIds.mapsubstitui cada entrada da lista (ou seja, a lista agora contém Strings em vez de Ints).

    Voltando à pergunta original.

    Como está agora, getProfileFromNetwork()é chamado sequentially . Se você quiser que isso aconteça concorrentemente , você precisa lançar uma nova coroutine para cada:

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

    asynclança uma nova coroutine no escopo fornecido por runBlocking, e awaitAll()espera que todas as coroutines terminem. Ele retorna uma lista de perfis, ou seja, um List<String>.

    Isso é repetido para cada novo valor de fluxo, então o perfil para o ID do usuário 0é recuperado três vezes no total. O perfil para o ID do usuário 2é recuperado apenas uma vez porque ocorre apenas na última emissão de fluxo.

    Como getProfileFromNetworké mais lento do que novos valores de fluxo são emitidos, você pode querer alternar de flowOfIds.mappara flowOfIds.mapLatest. Este cancela a transformação de fluxo anterior quando um novo valor de fluxo chega, então você não precisa esperar a transformação anterior terminar antes de iniciar a nova.

    Além disso, você deve considerar seriamente armazenar em cache os resultados de getProfileFromNetwork(), para não fazer repetidamente as mesmas solicitações de rede (como buscar 0o perfil do id do usuário três vezes). Isso é algo que deve ser feito dentro de getProfileFromNetwork(). Você provavelmente deve mover essa função para uma classe de repositório ou algo semelhante, para poder construir seu cache lá.

    • 1

relate perguntas