Flow для Kotlin: полное руководство с примерами

Введение в Flow: что это такое и зачем нужно?

Flow — это библиотека для работы с реактивными потоками данных (reactive streams) в языке программирования Kotlin. Она является частью экосистемы Kotlin Coroutines и предоставляет мощный, типобезопасный и асинхронный способ обработки последовательностей данных, которые генерируются с течением времени.

В отличие от обычных коллекций (List, Set), которые хранят все элементы в памяти сразу, Flow представляет собой холодный поток (cold stream). Это означает, что данные начинают генерироваться только тогда, когда появляется подписчик (collector). Flow идеально подходит для:

• Асинхронной обработки данных: Получение данных из сети, базы данных или файловой системы без блокировки основного потока.
• Обработки событий: Клики, движения мыши, обновления UI.
• Реактивного программирования: Создание цепочек преобразований данных (map, filter, flatMap) с автоматической обработкой ошибок и отменой.
• Интеграции с корутинами: Flow полностью совместим с Kotlin Coroutines, что позволяет легко комбинировать его с другими асинхронными операциями.

Зачем использовать Flow вместо старых подходов (например, RxJava или callback'ов)?

• Простота и лаконичность: API Flow минималистичен и интуитивно понятен для разработчиков на Kotlin.
• Безопасность: Типобезопасность на уровне компилятора.
• Интеграция с корутинами: Использует все преимущества structured concurrency (структурированной конкурентности).
• Производительность: Легковесный и эффективный, не создает лишних объектов.

Установка Flow

Flow является частью библиотеки kotlinx.coroutines. Для его использования вам нужно добавить зависимость в ваш файл build.gradle.kts (для Gradle) или pom.xml (для Maven).

Gradle (Kotlin DSL)

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
    // Для использования Flow в Android (с поддержкой Lifecycle)
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1")
}

Gradle (Groovy)

dependencies {
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1'
    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.8.1'
}

Maven

    org.jetbrains.kotlinx
    kotlinx-coroutines-core
    1.8.1

Важно: Убедитесь, что версия Kotlin и версия kotlinx-coroutines совместимы. Актуальную версию можно найти на GitHub.

Быстрый старт: первый Flow

Давайте создадим простой Flow, который эмитирует три числа с задержкой, и подпишемся на него.

import kotlinx.coroutines.*
import kotlinx.coroutines.flow.*

fun simpleFlow(): Flow<Int> = flow {
    println("Flow стартовал")
    for (i in 1..3) {
        delay(100) // Имитация асинхронной работы
        emit(i)    // Эмитируем значение
    }
}

fun main() = runBlocking {
    println("Главный поток: ${Thread.currentThread().name}")

    // Подписываемся на Flow (коллектим данные)
    simpleFlow().collect { value ->
        println("Получено: $value")
    }

    println("Flow завершен")
}

Вывод:

Главный поток: main
Flow стартовал
Получено: 1
Получено: 2
Получено: 3
Flow завершен

Объяснение:

• flow { ... } — создает холодный поток.
• emit() — функция, которая отправляет значение подписчику.
• collect { ... } — терминальная операция, которая запускает выполнение Flow и обрабатывает каждое эмитированное значение.
• delay() — неблокирующая задержка (корутина).

Основные методы и функции Flow

Рассмотрим 10 ключевых методов, которые вы будете использовать чаще всего. Каждый метод представлен с описанием, параметрами и рабочим примером.

1. flow { } — Создание потока

Что делает: Создает новый холодный асинхронный поток. Внутри блока вы можете вызывать emit() для отправки данных.

Параметры: Нет. Это функция-конструктор, которая принимает suspend-блок.

val numberFlow: Flow<Int> = flow {
    emit(1)
    delay(500)
    emit(2)
    emit(3)
}

2. collect — Терминальная операция (подписка)

Что делает: Запускает выполнение Flow и обрабатывает каждое эмитированное значение. Блокирует корутину до тех пор, пока Flow не завершится или не будет отменен.

Параметры: Принимает лямбда-функцию (suspend-блок), которая вызывается для каждого элемента.

runBlocking {
    flowOf("A", "B", "C").collect { value ->
        println("Получено: $value")
    }
}

3. map — Преобразование элементов

Что делает: Применяет заданную функцию к каждому элементу потока и возвращает новый поток с преобразованными значениями.

Параметры: transform: suspend (T) -> R — функция преобразования.

runBlocking {
    val numbers = (1..5).asFlow()
    numbers.map { it * it } // Возводим в квадрат
        .collect { println(it) } // Вывод: 1, 4, 9, 16, 25
}

4. filter — Фильтрация элементов

Что делает: Пропускает только те элементы, которые удовлетворяют заданному условию.

Параметры: predicate: suspend (T) -> Boolean — функция-предикат.

runBlocking {
    (1..10).asFlow()
        .filter { it % 2 == 0 } // Только четные
        .collect { println(it) } // Вывод: 2, 4, 6, 8, 10
}

5. flatMapConcat — Последовательное преобразование (конкатенация)

Что делает: Преобразует каждый элемент в новый Flow, а затем последовательно объединяет все полученные потоки в один. Второй поток начинается только после завершения первого.

Параметры: transform: suspend (T) -> Flow<R> — функция, возвращающая новый Flow.

runBlocking {
    flowOf("A", "B").flatMapConcat { letter ->
        flow {
            emit("$letter-1")
            delay(100)
            emit("$letter-2")
        }
    }.collect { println(it) }
    // Вывод: A-1, A-2, B-1, B-2
}

6. catch — Обработка ошибок

Что делает: Перехватывает исключения, возникшие вверх по потоку (до оператора catch), и позволяет выполнить альтернативную логику (например, эмитить значение по умолчанию или логировать ошибку).

Параметры: action: suspend FlowCollector<T>.(cause: Throwable) -> Unit — блок обработки ошибки.

runBlocking {
    flow {
        emit(1)
        throw RuntimeException("Ошибка!")
    }.catch { e ->
        println("Поймана ошибка: ${e.message}")
        emit(-1) // Эмитим значение по умолчанию
    }.collect { println(it) }
    // Вывод: 1, Поймана ошибка: Ошибка!, -1
}

7. onCompletion — Действие при завершении

Что делает: Выполняет заданное действие, когда Flow завершается (нормально или из-за ошибки). Позволяет узнать причину завершения.

Параметры: action: suspend FlowCollector<T>.(cause: Throwable?) -> Unit. Если cause == null — завершение нормальное, иначе — ошибка.

runBlocking {
    flowOf(1, 2, 3)
        .onCompletion { cause ->
            if (cause != null) {
                println("Завершено с ошибкой: ${cause.message}")
            } else {
                println("Завершено успешно")
            }
        }
        .collect { println(it) }
    // Вывод: 1, 2, 3, Завершено успешно
}

8. debounce — Дебаунс (подавление частых событий)

Что делает: Пропускает элементы, если между двумя последовательными эмиссиями прошло меньше заданного времени. Полезно для обработки поисковых запросов или кликов.

Параметры: timeoutMillis: Long — время ожидания в миллисекундах.

runBlocking {
    flow {
        emit("A")
        delay(50)
        emit("B")
        delay(200) // Пауза больше 150 мс
        emit("C")
    }.debounce(150)
        .collect { println(it) }
    // Вывод: B, C (A был отброшен, так как B пришел слишком быстро)
}

9. distinctUntilChanged — Удаление повторяющихся элементов подряд

Что делает: Пропускает элемент, только если он отличается от предыдущего. Избавляет от дубликатов, идущих подряд.

Параметры: Нет (использует стандартное сравнение equals).

runBlocking {
    flowOf(1, 1, 2, 2, 3, 1)
        .distinctUntilChanged()
        .collect { println(it) }
    // Вывод: 1, 2, 3, 1 (повторяющиеся 1 и 2 были удалены)
}

10. zip — Объединение двух потоков

Что делает: Объединяет два потока в один, попарно комбинируя их элементы. Завершается, когда завершается самый короткий из потоков.

Параметры: other: Flow<R> — второй поток; transform: suspend (T, R) -> V — функция объединения.

runBlocking {
    val numbers = flowOf(1, 2, 3)
    val letters = flowOf("A", "B", "C", "D") // D будет проигнорирован
    numbers.zip(letters) { num, letter -> "$num$letter" }
        .collect { println(it) }
    // Вывод: 1A, 2B, 3C
}

Реальный практический пример: Поиск с автодополнением

Представьте, что у нас есть поле поиска, и мы хотим отправлять запрос на сервер только после того, как пользователь перестал печатать (debounce), и игнорировать повторяющиеся запросы (distinctUntilChanged).

import kotlinx.coroutines.*
import kotlin