¿Cómo se pueden cambiar los hilos en Flow? ¿En qué…

Por defecto, Flow funciona en el hilo desde el cual fue iniciado. Puedes cambiar de hilo usando el operador flowOn.

import kotlinx.coroutines.CoroutineDispatcher // Para CoroutineDispatcher
import kotlinx.coroutines.Dispatchers // Para Dispatchers
import kotlinx.coroutines.flow.Flow // Para Flow
import kotlinx.coroutines.flow.flow // Para flow
import kotlinx.coroutines.flow.flowOn // Para flowOn
import kotlinx.coroutines.runBlocking // Para runBlocking

// Ejemplo de función que devuelve un Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow iniciado") // Registro del hilo donde se inicia el builder de flow
    for (i in 1..3) {
        emit(i) // Emitimos valores
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Cambiamos upstream (emitores) a Dispatchers.IO
        .collect { value ->
            logThread("Recogido $value") // Registro del hilo donde se recogen los valores
        }
}

// Función auxiliar para registrar el hilo
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

El operador flowOn afecta al hilo en el que se ejecutan los operadores antes de él en la cadena (incluyendo el flow builder). Los operadores después de flowOn se ejecutan en el hilo especificado en el argumento flowOn. Si hay varios flowOn en la cadena, cada uno afecta a la parte de la cadena antes de él.

Ejemplo con múltiples flowOn:

import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.runBlocking

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Builder de Flow iniciado") // Funcionará en el primer flowOn especificado
    emit("A")
    emit("B")
}.map {
    logThread("Mapeo $it") // Funcionará en el segundo flowOn especificado
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // El segundo flowOn afecta a map() y al builder de flow
.filter {
    logThread("Filtrando $it") // Funcionará en el hilo donde se llama a collect (por defecto main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // El primer flowOn afecta a map() y al builder de flow

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Recogiendo $value") // Funciona en el hilo runBlocking (main)
    }
}

fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

En este ejemplo:

El builder de flow y map funcionarán en Dispatchers.Default (porque es el último flowOn en la cadena antes de ellos).
filter funcionará en el hilo desde el cual se llama a collect, es decir, en el hilo donde se ejecuta runBlocking (normalmente Main en Android).

Es importante entender que flowOn crea un buffer intermedio y no es un "cambio transparente" de hilo.

¿Cómo se pueden cambiar los hilos en Flow? ¿En qué hilo funciona Flow?