Kotlin Flow — от основ до продакшена: практическое руководство

Kotlin Flow — одна из самых мощных и одновременно недопонятых абстракций в стандартной библиотеке. Каждый Kotlin-разработчик знает suspend-функции, но когда дело доходит до Flow — начинаются вопросы. Зачем Flow, если есть Sequence? Чем StateFlow отличается от LiveData? Что такое горячий и холодный поток?

За последний год я использовал Flow в production для WebSocket-стриминга, SSE-эндпоинтов, real-time уведомлений и data pipeline для аналитики. Рассказываю, как это работает на практике.

Холодные потоки: основы Flow

Flow — это холодный асинхронный поток. «Холодный» означает, что код внутри flow { } не выполняется, пока кто-нибудь не вызовет collect. Каждый collector получает данные с начала — это не «подписка на живой источник», а «запуск pipeline».

fun fetchProducts(categoryId: Long): Flow<Product> = flow {
    var page = 0
    do {
        val products = productRepository.findByCategory(categoryId, page, 50)
        if (products.isEmpty()) break
        products.forEach { emit(it) }
        page++
    } while (products.size == 50)
}.flowOn(Dispatchers.IO) // Переключаем контекст для блокирующих операций

Здесь Flow работает как lazy Sequence, но асинхронно. Каждый emit() приостанавливает выполнение до тех пор, пока consumer не обработает предыдущее значение. Это и есть базовый backpressure: producer не генерирует данные быстрее, чем consumer может их потребить.

Операторы трансформации

Flow предоставляет богатый набор операторов, похожих на RxJava/Reactor, но в идиоматичном Kotlin-стиле:

productRepository.streamAll()
    .filter { it.price > BigDecimal.ZERO }
    .map { it.toDto() }
    .catch { e ->
        logger.error("Error streaming products", e)
        emit(ProductDto.empty()) // Fallback-значение
    }
    .onStart { logger.info("Starting product stream") }
    .onCompletion { cause ->
        if (cause == null) logger.info("Stream completed normally")
        else logger.error("Stream completed with error", cause)
    }
    .collect { productDto ->
        sendToKafka(productDto)
    }

Важно понимать: каждый оператор создаёт новый Flow. Операторы ленивые — они не запускают выполнение. Только collect (или терминальный оператор вроде toList(), first(), fold()) инициирует выполнение всей цепочки.

Критический нюанс — catch перехватывает исключения только вверх по потоку. Если ошибка произойдёт в операторе после catch, она не будет перехвачена. Правильный паттерн — ставить catch в конце цепочки, перед терминальным оператором.

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

Холодные потоки перезапускаются для каждого collector. Но что, если вам нужен один источник данных, который транслируется нескольким подписчикам? Например, WebSocket-соединение с ценами акций — вы не хотите открывать новое соединение для каждого подписчика.

Тут входят горячие потоки: SharedFlow и StateFlow.

// SharedFlow — для событий (неизменяемые значения, multiple consumers)
class OrderEventBus {
    private val _events = MutableSharedFlow<OrderEvent>(
        replay = 0,           // Не хранить прошлые события
        extraBufferCapacity = 64, // Буфер для быстрых продюсеров
        onBufferOverflow = BufferOverflow.SUSPEND // Или DROP_OLDEST для fire-and-forget
    )
    val events: SharedFlow<OrderEvent> = _events.asSharedFlow()

    suspend fun publish(event: OrderEvent) {
        _events.emit(event)
    }
}

// StateFlow — для состояния (одно текущее значение, multiple observers)
class CartViewModel {
    private val _cartState = MutableStateFlow(CartState(emptyList(), BigDecimal.ZERO))
    val cartState: StateFlow<CartState> = _cartState.asStateFlow()

    fun addItem(item: CartItem) {
        _cartState.update { current ->
            current.copy(
                items = current.items + item,
                totalPrice = current.totalPrice + item.price
            )
        }
    }
}

StateFlow — это специализированный SharedFlow с replay = 1 и conflation. Он всегда имеет актуальное значение (.value) и идеально подходит для хранения состояния (замена LiveData, BehaviorSubject из RxJava). Новые подписчики мгновенно получают текущее значение.

SharedFlow — для событий (однократных или повторяющихся). Подписчик получает только те события, которые произошли после подписки. Если replay = 0, подписчик пропустит все прошлые события.

Классическая ошибка: SharedFlow как StateFlow

Частая ошибка — использование SharedFlow(replay = 1) вместо StateFlow для состояния. Разница в поведении при эмиссии: StateFlow заменяет одинаковые значения (если .value не изменилось, подписчики не уведомляются), а SharedFlow всегда уведомляет.

// НЕПРАВИЛЬНО: SharedFlow для состояния
private val _count = MutableSharedFlow<Int>(replay = 1)
// Если emit(5) -> emit(5), подписчики получат ДВА события

// ПРАВИЛЬНО: StateFlow для состояния
private val _count = MutableStateFlow(0)
// Если emit(5) -> emit(5), подписчики получат только одно (значение не изменилось)
// Важно: StateFlow сравнивает через equals(), не через ===

Backpressure: стратегии

Backpressure в Flow реализуется неявно через suspend. Producer приостанавливается на emit(), если consumer не готов принимать данные. Но есть сценарии, где нужна явная стратегия:

// conflate: пропускать промежуточные значения, оставляя последнее
sensorDataStream()
    .conflate()
    .collect { latestReading ->
        updateDashboard(latestReading)
    }

// buffer: буферизировать N элементов, consumer читает с задержкой
sensorDataStream()
    .buffer(capacity = 100)
    .collect { reading ->
        writeToDatabase(reading) // Медленный consumer
    }

// collectLatest: отменять предыдущую обработку при новом значении
searchResultsFlow()
    .collectLatest { query ->
        val results = searchService.search(query) // Долгий поиск
        updateUI(results)
    }

collectLatest — мощный оператор для UI-сценариев. Пользователь вводит текст в поиск, вы запускаете API-запрос. Если пользователь ввёл новый символ до завершения предыдущего запроса — предыдущий запрос отменяется (через Job cancellation корутин). Это debounce + отмена в одном операторе.

Error handling в продакшене

Обработка ошибок — самая частая проблема при работе с Flow в production. Стратегия зависит от сценария:

// 1. Retry с экспоненциальной задержкой для временных сбоев
externalApiStream()
    .retry(3) {
        if (it is IOException) {
            delay(it.retryDelay()) // Кастомная экспоненциальная задержка
            true // Повторяем
        } else false // Не повторяем для других ошибок
    }
    .catch { e ->
        logger.error("Stream failed after retries", e)
        emit(FallbackData)
    }

// 2. Перезапуск с последнего успешного элемента
dataPipeline()
    .catch { e ->
        logger.error("Pipeline error, restarting from last offset", e)
        emitAll(dataPipeline(fromOffset = lastProcessedOffset))
    }

// 3. Изоляция ошибок по элементам
itemStream()
    .map { item ->
        try {
            processItem(item)
        } catch (e: Exception) {
            logger.error("Failed to process item ${item.id}", e)
            ItemResult.Failed(item.id, e.message)
        }
    }

Ключевое правило: catch и retry — это операторы вверх по потоку. Они перехватывают ошибки от операторов, которые стоят до них. Ставьте их после источника данных, перед бизнес-трансформациями.

Real-time: WebSocket + Flow

Один из лучших use-кейсов для Flow — WebSocket-стриминг. Реальный пример из продакшена — live-обновление цен в маркетплейсе:

class PriceStreamService(
    private val websocketClient: WebSocketClient
) {
    private val _prices = MutableSharedFlow<PriceUpdate>(
        replay = 0,
        extraBufferCapacity = 256,
        onBufferOverflow = BufferOverflow.DROP_OLDEST
    )

    val prices: SharedFlow<PriceUpdate> = _prices.asSharedFlow()

    suspend fun start() {
        websocketClient.connect("wss://prices.example.com/stream")
            .receiveAsFlow()
            .map { frame ->
                json.decodeFromString<PriceUpdate>(frame.readText())
            }
            .catch { e ->
                logger.error("WebSocket error, reconnecting in 5s", e)
                delay(5000)
                start() // Рекурсивное переподключение
            }
            .collect { priceUpdate ->
                _prices.emit(priceUpdate)
            }
    }
}

// В Ktor endpoint
get("/api/v1/prices/stream") {
    call.respond(
        contentType = ContentType.Text.EventStream,
        status = HttpStatusCode.OK
    ) {
        priceStreamService.prices
            .map { update ->
                "data: ${json.encodeToString(update)}\n\n"
            }
            .catch { emit("event: error\ndata: ${it.message}\n\n") }
            .collect { chunk -> call.response.out.write(chunk.toByteArray()) }
    }
}

Этот паттерн — WebSocket-источник + SharedFlow для broadcast + SSE для фронтенд-клиентов. На продакшене он держит 10 000 одновременных SSE-подключений с задержкой обновления менее 50 мс.

Мой вердикт

Flow — мощная абстракция, но не серебряная пуля. Используйте холодные Flow для pipeline-обработки данных (пагинация, batch-обработка, ETL). Используйте StateFlow для хранения состояния (состояние вьюмодели, конфигурация). Используйте SharedFlow для широковещания событий (нотификации, WebSocket-события, pub/sub).

Главный совет: не превращайте всё в Flow. Если вам нужно получить одно значение из БД — используйте suspend-функцию. Flow нужен тогда, когда данных много, они приходят асинхронно, и вам нужна обработка в стиле потоков. Использование Flow там, где достаточно suspend, — это антипаттерн, который усложняет код без выгоды.