Реактивное программирование в Kotlin: Flow и корутины |…

Реактивное программирование в Kotlin: мощь kotlinx.coroutines и Flow

Реактивное программирование — это парадигма, ориентированная на асинхронную обработку потоков данных и распространение изменений. В экосистеме Kotlin основным инструментом для реализации реактивных паттернов является библиотека kotlinx.coroutines, а точнее её компонент Flow. В отличие от RxJava, Flow тесно интегрирован с корутинами, что делает код более лаконичным и безопасным.

Основные концепции Flow

Flow — это холодный асинхронный поток данных, который эмитирует значения последовательно. Он поддерживает:

Холодные потоки — данные начинают генерироваться только при наличии подписчика (терминального оператора).
Контекст выполнения — поток может переключаться между разными диспетчерами корутин.
Отмену через корутины — отмена родительской корутины автоматически завершает Flow.

Установка и подключение

Для использования Flow добавьте зависимость в build.gradle.kts:

dependencies {
    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.8.1")
}

Создание и потребление Flow

Создадим простой поток чисел с задержкой:

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

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

fun main() = runBlocking {
    simpleFlow().collect { value ->
        println("Получено: $value")
    }
}

В этом примере flow { ... } создаёт холодный поток, а collect — терминальный оператор, запускающий выполнение.

Трансформация данных с помощью операторов

Flow предоставляет богатый набор операторов, аналогичных RxJava:

map — преобразует каждое значение.
filter — фильтрует значения по условию.
flatMapConcat — последовательно объединяет потоки.
catch — обрабатывает исключения внутри потока.

Пример с операторами:

fun transformFlow(): Flow<String> = flow {
    emit("Kotlin")
    emit("Java")
    emit("Python")
}.map { it.lowercase() }
 .filter { it.startsWith("k") }

fun main() = runBlocking {
    transformFlow().collect { println(it) }
    // Вывод: kotlin
}

Обработка ошибок и завершения

Flow позволяет элегантно обрабатывать исключения с помощью catch и выполнять завершающие действия через onCompletion:

fun errorProneFlow(): Flow<Int> = flow {
    emit(1)
    emit(2)
    throw RuntimeException("Ошибка!")
    emit(3) // не выполнится
}

fun main() = runBlocking {
    errorProneFlow()
        .catch { e -> println("Поймана ошибка: ${e.message}") }
        .onCompletion { println("Поток завершён") }
        .collect { println(it) }
}
// Вывод: 1, 2, Поймана ошибка: Ошибка!, Поток завершён

Горячие потоки: StateFlow и SharedFlow

В отличие от холодного Flow, StateFlow и SharedFlow являются горячими потоками, которые хранят состояние и могут иметь несколько подписчиков.

StateFlow

Используется для хранения и наблюдения за состоянием (например, в ViewModel):

class CounterViewModel {
    private val _counter = MutableStateFlow(0)
    val counter: StateFlow<Int> = _counter.asStateFlow()

    fun increment() {
        _counter.value++
    }
}

// В Activity или Fragment
lifecycleScope.launch {
    viewModel.counter.collect { value ->
        textView.text = "Счёт: $value"
    }
}

SharedFlow

Подходит для событий, которые должны быть обработаны только один раз (например, навигационные команды):

class EventBus {
    private val _events = MutableSharedFlow<String>(replay = 0)
    val events: SharedFlow<String> = _events

    suspend fun sendEvent(event: String) {
        _events.emit(event)
    }
}

suspend fun consumeEvents(bus: EventBus) {
    bus.events.collect { event ->
        println("Событие: $event")
    }
}