0Pricing
Kotlin Academy · Aula

flowOn e buffer: contexto e contrapressão

Altere o contexto de emissão com flowOn e armazene emissões em buffer para lidar com a contrapressão.

flowOn e buffer: contexto e contrapressão é uma aula grátis de Kotlin Academy no CoddyKit. Esta é a aula 4 de 4. Você pode ler a aula completa abaixo gratuitamente — depois pratica ao vivo no navegador com um editor de código integrado e um tutor de IA 24/7. Faz parte do caminho de aprendizado de Kotlin Academy, e seu progresso é sincronizado entre a web e o app CoddyKit. O curso de Kotlin Academy inclui 4 aulas no total.

Contexto do fluxo

Por padrão, um fluxo é executado no contexto da corrotina que chama collect. O despachador do produtor e o do consumidor são iguais, a menos que você os altere.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit on: ${Thread.currentThread().name}")
        emit(1)
    }.collect {
        println("collect on: ${Thread.currentThread().name}")
    }
}

flowOn altera o contexto upstream

flowOn(dispatcher) executa o fluxo upstream (tudo o que está acima dele na cadeia) no despachador especificado, enquanto a coleta permanece no despachador do chamador.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        println("emit: ${Thread.currentThread().name}")
        emit(1)
    }.map {
        println("map: ${Thread.currentThread().name}")
        it * 2
    }.flowOn(Dispatchers.Default)  // above runs on Default
    .collect {
        println("collect: ${Thread.currentThread().name}")
    }
}

Vários flowOn em uma cadeia

Você pode usar flowOn várias vezes. Cada ocorrência afeta os operadores diretamente acima dela, até o flowOn anterior.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow { emit(1) }
        .map { it + 1 }.flowOn(Dispatchers.IO)      // map runs on IO
        .map { it * 2 }.flowOn(Dispatchers.Default)  // this map runs on Default
        .collect { println("Result: $it") }           // collect on Main (runBlocking)
}

O problema da pressão reversa

Quando o produtor emite mais rapidamente do que o coletor consegue processar, os valores se acumulam em uma fila. Sem armazenamento em buffer, isso faz o produtor esperar.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i ->
            delay(100)  // fast producer
            emit(i)
        }
    }.collect {
        delay(300)      // slow consumer
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

Operador buffer()

buffer() executa o produtor e o consumidor simultaneamente em corrotinas separadas, armazenando os valores emitidos em um canal. O produtor não espera pelo consumidor.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    val time = System.currentTimeMillis()
    flow {
        repeat(3) { i -> delay(100); emit(i) }
    }.buffer()  // producer and consumer run concurrently
    .collect {
        delay(300)
        println("Got $it in ${System.currentTimeMillis() - time}ms")
    }
}

Capacidade de buffer

buffer(capacity) define o tamanho do buffer do canal. Quando o buffer está cheio, o produtor é suspenso (pressão reversa). A capacidade padrão é 64.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.channels.Channel
fun main() = runBlocking {
    flow { repeat(5) { emit(it) } }
        .buffer(Channel.RENDEZVOUS)    // 0: producer waits
        // .buffer(Channel.BUFFERED)   // default: 64
        // .buffer(Channel.UNLIMITED)  // unbounded
        .collect { delay(50); println(it) }
}

conflate() para manter apenas o mais recente

conflate() descarta valores intermediários quando o coletor está lento, mantendo apenas a emissão mais recente. É útil para o estado da interface.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        repeat(5) { i -> emit(i); delay(50) }
    }.conflate()
    .collect { i ->
        delay(150)
        println("Collected: $i")  // skips some values
    }
}

collectLatest para coletores lentos

collectLatest cancela o bloco de coleta atual quando um novo valor chega e o reinicia com o valor mais recente.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun main() = runBlocking {
    flow {
        emit(1); delay(50)
        emit(2); delay(50)
        emit(3)
    }.collectLatest { value ->
        println("Processing $value")
        delay(100)  // gets cancelled if new value arrives
        println("Done $value")  // only prints for last value
    }
}

Padrão flowOn + buffer

Combine flowOn e buffer: execute o produtor em IO (rede/disco), armazene os resultados em buffer e faça a coleta no despachador Principal.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun fetchItems(): Flow<String> = flow {
    repeat(3) { i ->
        delay(100)  // simulate IO
        emit("item-$i")
    }
}.flowOn(Dispatchers.IO).buffer(10)
fun main() = runBlocking {
    fetchItems().collect { println("UI: $it") }
}

channelFlow para produtores simultâneos

channelFlow cria um fluxo apoiado por um canal, permitindo que várias corrotinas emitam simultaneamente dentro do construtor.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
fun concurrentFlow(): Flow<Int> = channelFlow {
    launch { send(1) }
    launch { send(2) }
    launch { send(3) }
}
fun main() = runBlocking {
    concurrentFlow().collect { println(it) }
}

Escolhendo a estratégia correta

Resumo: flowOn para troca de contexto, buffer para vazão, conflate para interfaces que mantêm apenas o valor mais recente e collectLatest para cancelar processamento obsoleto.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*
// Guidelines:
// CPU-heavy production  -> flowOn(Dispatchers.Default)
// IO-heavy production   -> flowOn(Dispatchers.IO)
// Slow consumer         -> buffer()
// UI state updates      -> conflate() or StateFlow
// Search/autocomplete   -> collectLatest or flatMapLatest
fun main() = runBlocking { println("Choose the right strategy!") }

Verificação rápida

O que flowOn afeta em uma cadeia de fluxos?

Recapitulação

flowOn move o trabalho upstream para outro despachador. buffer desacopla o produtor e o consumidor para aumentar a vazão. conflate descarta valores intermediários. collectLatest cancela o processamento lento quando chegam novos valores.

Perguntas Frequentes

A aula “flowOn e buffer: contexto e contrapressão” é grátis?

Sim — o texto completo de “flowOn e buffer: contexto e contrapressão” é grátis para ler aqui na web. Para praticá-la interativamente (um editor de código integrado e um tutor de IA 24/7) e desbloquear o restante do curso de Kotlin Academy, atualize para CoddyKit PRO. O curso de Kotlin Academy inclui 4 aulas no total.

O que vou aprender em “flowOn e buffer: contexto e contrapressão”?

Altere o contexto de emissão com flowOn e armazene emissões em buffer para lidar com a contrapressão. Você pratica Kotlin Academy com código prático que executa diretamente no navegador, e um tutor de IA 24/7 responde suas dúvidas enquanto trabalha na aula.

Preciso ter experiência prévia para começar Kotlin Academy?

Nenhuma experiência prévia é necessária. Kotlin Academy no CoddyKit é estruturado para alunos iniciantes até avançados, então você pode começar aqui ou desde o início e aprender no seu ritmo. Esta é a aula 4 de 4.

Quanto tempo leva a aula “flowOn e buffer: contexto e contrapressão”?

A maioria das aulas CoddyKit leva cerca de 5–10 minutos. Cada uma é compacta e interativa, então você faz progresso constante e retoma exatamente de onde parou entre web e app.

Posso escrever e executar código nesta aula de Kotlin Academy?

Sim. Cada aula de Kotlin Academy inclui um editor de código integrado, então você escreve e executa código real direto no navegador e recebe feedback de IA instantaneamente — nenhuma configuração local necessária.

Todas as aulas deste curso

  1. Operadores de Flow: map, filter, transform e take
  2. catch e onCompletion: tratamento de erros em Flow
  3. combine e zip: mesclando vários Flows
  4. flowOn e buffer: contexto e contrapressão
← Voltar para Kotlin Academy