Flow в Kotlin

ОБЯЗАТЕЛЬНОДЛЯ НОВИЧКОВ

Разработчику Архитектору

Flow в Kotlin

Корутина выполняет одну задачу и возвращает один результат. Flow — последовательность значений во времени: пришла новая строка из БД, обновился счётчик, прилетела порция из сети.

Типичные места:

• Room — observeAll(): Flow<List<Note>>;
• Ktor — стриминг ответов;
• Compose — collectAsState из StateFlow.

База: корутины · синхронные коллекции: 225.

---

List, Sequence и Flow — сравнение

	List	Sequence	Flow
Когда данные	Всё сразу в памяти	Лениво, в одном потоке	Лениво + suspend
Сеть / БД	Нужно самому загрузить в List	То же	Может эмитить по мере готовности
Повторный запуск	Тот же объект	Новый проход итератора	Холодный Flow — заново при новом collect
Контекст	Обычный код	Обычный код	Корутины

---

Словарь терминов

Термин	Простыми словами
emit	«Выдать» следующее значение подписчикам Flow.
collect	Подписаться и обрабатывать каждое значение (suspend-цикл).
Холодный Flow	Начинает работу, когда появился первый подписчик.
Горячий поток	Живёт сам по себе; подписчики получают текущее состояние (StateFlow).
Upstream / downstream	Продюсер → операторы → collect.
Backpressure	Продюсер быстрее потребителя — буферы (buffer).

---

Холодный Flow

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

fun counter(): Flow<Int> = flow {
    for (i in 1..5) {
        delay(200)
        emit(i)
    }
}

suspend fun main() {
    counter().collect { println(it) }
}

Разбор:

Часть	Смысл
flow { }	Билдер: внутри можно emit и вызывать suspend-функции.
delay(200)	Пауза между числами; поток не блокируется.
collect { }	Запускает выполнение; без collect холодный Flow ничего не делает.

Два подряд counter().collect { } выполнят цикл дважды — это норма для холодного потока.

---

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

Экрану нужно одно актуальное состояние («загрузка» / «список» / «ошибка»):

private val _uiState = MutableStateFlow(UiState.Loading)
val uiState: StateFlow<UiState> = _uiState.asStateFlow()

fun onDataLoaded(data: List<Item>) {
    _uiState.value = UiState.Success(data)
}

Тип	Когда
StateFlow	Состояние UI, настройки — всегда есть «последнее значение»
SharedFlow	События: «показать Snackbar», навигация один раз
channelFlow	Несколько продюсеров пишут в один Flow

В Compose: uiState.collectAsStateWithLifecycle() — подписка с учётом жизненного цикла.

---

Операторы (как у коллекций, но suspend)

repository.observeOrders()
    .map { orders -> orders.filter { it.isOpen } }
    .distinctUntilChanged()
    .catch { e -> emit(emptyList()); log(e) }
    .flowOn(Dispatchers.IO)
    .collect { openOrders -> render(openOrders) }

Оператор	Роль
map	Преобразовать каждое значение
filter	Отфильтровать
distinctUntilChanged	Пропустить, если новое equals предыдущему (меньше перерисовок UI)
catch	Поймать ошибку upstream, можно emit запасное значение
flowOn(Dispatchers.IO)	Тяжёлую работу upstream выполнять на IO
debounce	Подождать паузу ввода (поиск по буквам)
flatMapLatest	При новом значении отменить предыдущую вложенную работу

flatMapLatest типичен для поиска: пользователь напечатал «к» → «ко» → «кот»; запрос за «к» отменяется, когда пришло «ко».

---

Объединение потоков

val merged = merge(flowA, flowB)

val combined = combine(tempFlow, humidityFlow) { t, h ->
    Weather(t, h)
}

• merge — пришло значение из любого источника.
• combine — ждёт обновления каждого участника, строит результат из последних пар.

---

Отмена

collect — suspend-функция. При отмене viewModelScope (уход с экрана) цепочка останавливается, upstream получает CancellationException.

Внутри flow { } блокирующий код без withContext(Dispatchers.IO) может занять поток диспетчера — переносите тяжёлую работу на IO.

---

Тестирование

@Test
fun emitsFirstThree() = runTest {
    val items = counter().take(3).toList()
    assertEquals(listOf(1, 2, 3), items)
}

Для пошаговой проверки событий по мере прихода удобна библиотека Turbine — см. экосистему.

---

Частые ошибки

Симптом	Причина
Flow «молчит»	Нет collect у холодного Flow
Двойной запрос к API	Два collect без shareIn / кэша
UI дёргается	Нет distinctUntilChanged на частых обновлениях
Зависание	Блокирующий код в flow { } на Main

---

Дальше

Корутины · Room + Flow · Compose · тесты

---