По умолчанию Flow работает на том потоке, с которого был запущен. Переключать потоки можно с помощью оператора flowOn.

import kotlinx.coroutines.CoroutineDispatcher // Для CoroutineDispatcher
import kotlinx.coroutines.Dispatchers // Для Dispatchers
import kotlinx.coroutines.flow.Flow // Для Flow
import kotlinx.coroutines.flow.flow // Для flow
import kotlinx.coroutines.flow.flowOn // Для flowOn
import kotlinx.coroutines.runBlocking // Для runBlocking

// Пример функции, возвращающей Flow
fun simpleFlow(): Flow<Int> = flow {
    logThread("Flow started") // Логирование потока, где запускается flow builder
    for (i in 1..3) {
        emit(i) // Эмитим значения
    }
}

fun main() = runBlocking {
    simpleFlow()
        .flowOn(Dispatchers.IO) // Переключаем upstream (emitters) на Dispatchers.IO
        .collect { value ->
            logThread("Collected $value") // Логирование потока, где собираются значения
        }
}

// Вспомогательная функция для логирования потока
fun logThread(msg: String) {
    println("[$msg] [${Thread.currentThread().name}]")
}

Оператор flowOn влияет на поток, на котором выполняются операторы до него в цепочке (включая flow builder). Операторы после flowOn выполняются на потоке, указанном в аргументе flowOn. Если в цепочке несколько flowOn, каждый из них влияет на часть цепочки перед ним.

Пример с несколькими flowOn:

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

fun simpleFlowWithMultipleFlowOn(): Flow<String> = flow {
    logThread("Flow builder started") // Будет работать на первом specified flowOn
    emit("A")
    emit("B")
}.map {
    logThread("Mapping $it") // Будет работать на втором specified flowOn
    it.toLowerCase()
}.flowOn(Dispatchers.Default) // Второй flowOn влияет на map() и flow builder
.filter {
    logThread("Filtering $it") // Будет работать на потоке collect (по умолчанию main/runBlocking)
    true
}.flowOn(Dispatchers.IO) // Первый flowOn влияет на map() и flow builder

fun main() = runBlocking {
    simpleFlowWithMultipleFlowOn().collect { value ->
        logThread("Collecting $value") // Работает на потоке runBlocking (main)
    }
}

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

В этом примере:

flow builder и map будут работать на Dispatchers.Default (потому что это последний flowOn в цепочке перед ними).
filter будет работать на потоке, с которого вызывается collect, то есть на потоке, где runBlocking выполняется (обычно Main в Android).

Важно понимать, что flowOn создает промежуточный буфер и не является "прозрачным" переключением.

Как можно переключать потоки в Flow? На каком потоке работает Flow?