Flow, StateFlow и SharedFlow: урок по Kotlin к собеседованию

Flow это асинхронный поток значений на корутинах. В Android на нём построены UI-состояние, Room, DataStore и Paging. На собесе спросят про холодные и горячие потоки, разницу StateFlow и SharedFlow, flowOn, stateIn и операторы flatMapLatest, debounce, combine.

Холодный Flow

fun ticks(): Flow<Int> = flow {
    for (i in 1..3) {
        delay(100)
        emit(i)          // отдать значение подписчику
    }
}

scope.launch {
    ticks()
        .map { it * 10 }
        .filter { it > 10 }
        .collect { println(it) }   // 20, 30
}

Холодный: код внутри flow { } не выполняется, пока нет collect, и выполняется заново для каждого подписчика. Два collect значит два независимых запуска (два сетевых запроса, если там запрос).

collect это suspend-функция: она не вернётся, пока поток не закончится. Поэтому два бесконечных collect подряд в одной корутине не работают: второй никогда не начнётся. Каждый collect в своём launch.

Операторы map, filter, onEach, take, distinctUntilChanged промежуточные и тоже ленивые. Терминальные: collect, first, toList, single, launchIn(scope).

Контекст: flowOn

Flow сохраняет контекст: код flow { } выполняется в контексте того, кто вызвал collect. Нельзя внутри flow { } делать withContext(IO) { emit(x) }, будет IllegalStateException. Правильно flowOn:

repository.observe()
    .map { heavyParse(it) }
    .flowOn(Dispatchers.Default)   // всё выше выполняется на Default
    .collect { render(it) }        // а это в контексте коллектора (Main)

flowOn влияет только на операторы выше себя (upstream) и добавляет буфер между потоками.

Буферизация и скорость

Если коллектор медленный, по умолчанию производитель ждёт его на каждом элементе. Есть варианты:

buffer(): производитель работает вперёд, значения копятся в буфере;
conflate(): медленный коллектор получает только последнее значение, промежуточные пропускаются;
collectLatest { }: при новом значении текущая обработка отменяется.

Операторы, которые спрашивают

combine(a, b) { x, y -> }: новое значение при изменении любого источника, берёт последние из каждого;
zip: пары по порядку, ждёт оба;
flatMapLatest { }: на каждое значение запускает новый поток, предыдущий отменяет. Классика для поиска;
flatMapConcat (по очереди) и flatMapMerge (параллельно);
debounce(300): пропустить значение, только если после него была пауза;
catch { }: перехватить ошибку выше по цепочке, retry/retryWhen для повторов;
onStart, onCompletion.

queryFlow
    .debounce(300)
    .filter { it.length >= 2 }
    .distinctUntilChanged()
    .flatMapLatest { q -> repository.search(q) }
    .catch { emit(emptyList()) }

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

Горячий поток существует независимо от подписчиков, и все подписчики делят одни и те же значения.

StateFlow хранит одно текущее значение (value), новым подписчикам сразу отдаёт его. Значение по умолчанию обязательно. Одинаковые подряд значения не эмитит (сравнение через equals). Это замена LiveData для состояния UI:

class ProfileViewModel : ViewModel() {
    private val _state = MutableStateFlow<UiState>(UiState.Loading)
    val state: StateFlow<UiState> = _state.asStateFlow()

    fun load() = viewModelScope.launch {
        _state.update { UiState.Content(repo.profile()) }
    }
}

update { } атомарно меняет значение на основе текущего, без гонок, в отличие от _state.value = _state.value.copy(...).

SharedFlow не обязан иметь значение, настраивается: replay (сколько последних значений получит новый подписчик), extraBufferCapacity, onBufferOverflow. Подходит для событий. MutableSharedFlow() без параметров: без replay и без буфера, emit ждёт, пока все подписчики заберут значение, а если подписчиков нет, значение просто теряется.

Одноразовые события UI (показать снекбар, навигация) часто делают через Channel(Channel.BUFFERED).receiveAsFlow(): значение не теряется без подписчика и доставляется ровно одному. Многие команды сейчас предпочитают моделировать события как часть состояния.

Превращение холодного в горячий

val users: StateFlow<List<User>> = repository.observeUsers()
    .stateIn(viewModelScope, SharingStarted.WhileSubscribed(5_000), emptyList())

stateIn и shareIn запускают один upstream на всех подписчиков. WhileSubscribed(5000) останавливает upstream через 5 секунд после ухода последнего подписчика, этого хватает, чтобы пережить поворот экрана без перезапуска запроса. Eagerly стартует сразу, Lazily при первом подписчике и не останавливается.

Сбор в Android

Собирать нужно с учётом жизненного цикла, иначе поток работает в фоне:

lifecycleScope.launch {
    repeatOnLifecycle(Lifecycle.State.STARTED) {
        viewModel.state.collect { render(it) }
    }
}
// Compose:
val state by viewModel.state.collectAsStateWithLifecycle()

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

Ждать, что холодный Flow выполнится один раз на всех подписчиков.
withContext внутри flow { } вместо flowOn.
Два бесконечных collect подряд в одной корутине.
События через StateFlow: повторное значение не придёт, а после поворота событие покажется снова.
stateIn(scope, Eagerly, ...) для тяжёлых запросов, которые нужны только видимому экрану.
catch ниже коллектора: он ловит только ошибки upstream.

Что спросят на собесе

Холодный и горячий поток: в чём разница?
StateFlow против SharedFlow против LiveData.
Что делает flowOn и почему нельзя withContext внутри flow?
flatMapLatest, combine, zip, debounce: когда что?
Что такое stateIn и WhileSubscribed(5000)?
Как собирать Flow в Activity и Compose без утечек?

Flow, StateFlow и SharedFlow за 8 минут