В каких диспатчерах будет запущен Flow: map, filter, single

Движение данных Flow и диспатчеры операторов

В Kotlin Flow — это реактивный поток данных, построенный на корутинах. Ключевой принцип: операторы map, filter, single и большинство других промежуточных операторов не изменяют контекст корутины (диспатчер), в котором выполняется код их лямбда-выражений. Они наследуют контекст от вышестоящего потока.

Основное правило: контекст сохраняется

Промежуточные операторы выполняют свои transform-функции (лямбды) в том же диспатчере, в котором был запущен код оператора-коллектора (например, collect). Это происходит потому, что эти операторы по умолчанию не являются операторами с изменением контекста, такими как flowOn.

fun main() = runBlocking {
    val flow = flow {
        println("Emitting in ${Thread.currentThread().name}") // #1
        emit(1)
    }
        .map {
            println("Mapping in ${Thread.currentThread().name}") // #2
            it * 2
        }
        .filter {
            println("Filtering in ${Thread.currentThread().name}") // #3
            it > 0
        }

    flow
        .flowOn(Dispatchers.IO) // Меняет контекст для ВСЕГО, что ВЫШЕ
        .collect {
            println("Collecting in ${Thread.currentThread().name}") // #4
        }
}

Вывод (примерный):

• Emitting in DefaultDispatcher-worker-1 — из-за flowOn
• Mapping in DefaultDispatcher-worker-1 — унаследовал от эмиттера
• Filtering in DefaultDispatcher-worker-1 — унаследовал от вышестоящего
• Collecting in main — коллектор выполняется в runBlocking

Ключевые моменты и исключения

1. Оператор flowOn — это единственный стандартный оператор, который явно меняет контекст для всех операторов выше по цепочке. В примере выше flowOn(Dispatchers.IO) переключает эмиттер и все промежуточные операторы до себя на Dispatchers.IO, но не влияет на коллектор и операторы после него.
2. Коллектор (collect, single, toList) запускается в контексте родительской корутины. Если коллектор вызывается без явного launch или withContext, он будет блокировать текущую корутину.
3. Оператор single — это терминальный оператор, как и collect. Он также выполняется в контексте корутины-коллектора, но имеет важную особенность: он ожидает завершения потока и возврата единственного элемента. Если в потоке нет элементов или их больше одного, он выбросит исключение.
4. Операторы с собственной асинхронностью — например, callbackFlow или channelFlow — внутренняя логика эмита может использовать свои собственные диспатчеры, но трансформации (map, filter) будут применены в контексте, установленном для collect или flowOn.

Практическое правило для отладки

Чтобы понять, в каком потоке выполняется код внутри map или filter, достаточно посмотреть:

• Был ли применен flowOn выше по цепочке.
• В каком диспатчере запущен коллектор.

viewModelScope.launch(Dispatchers.Main.immediate) {
    repository.dataFlow
        .map { it * 2 } // Выполнится в IO из-за flowOn выше
        .flowOn(Dispatchers.IO)
        .collect {
            // Выполнится в Main, т.к. viewModelScope
            // по умолчанию использует Main-диспатчер для Android UI
        }
}

Вывод: По умолчанию map, filter, single не меняют диспатчер. Их контекст определяется коллектором и оператором flowOn, расположенным выше в цепочке. Это обеспечивает предсказуемость и позволяет управлять многопоточностью централизованно.