Мобилка

Flow: поток значений во времени

Прочитано0%

Квизы0/6

Тренажёрыне пройден0/1

Зачем это нужно
Одно значение и поток значений
Первый поток: flow и collect
Холодный поток: конвейер, который стоит
Операторы: map и filter
collect работает только в корутине
Ошибки внутри потока: catch и onCompletion
StateFlow: у состояния всегда есть значение
Кто пишет и кто читает
SharedFlow: событие вместо состояния
Где это встретится в приложении
Типичные ошибки и их настоящие сообщения
Слова и символы главы
Тренажёр: набрать вслепую
Что дальше
Углубиться
Куда дальше: проверенные ресурсы

Зачем это нужно

Прошлая глава закончилась обещанием. Корутина умеет дождаться одного ответа: попросили баланс — получили число, корутина закончилась. Так устроена suspend-функция: один вызов, одно значение, конец.

Но экран почти никогда не живёт одним значением. Список сообщений дополняется. Поле поиска шлёт запросы по мере ввода. Состояние экрана меняется: сначала «загрузка», потом «вот список», потом «ошибка, попробуйте ещё». Одним ответом это не описать — нужна последовательность значений, растянутая во времени.

Для этого в той же библиотеке корутин есть Flow. Это не отдельная технология и не замена корутинам: Flow построен на них же, и всё, что вы знаете про suspend, delay и отмену, продолжает работать.

Разница ровно одна: suspend-функция возвращает значение, а Flow — обещание присылать значения, пока его слушают.

Эта глава — фундамент для всего, что будет дальше в треке. База отдаёт данные потоком. Экран получает состояние потоком. Кэш, который переживёт отсутствие сети, собирается из двух потоков. Без Flow ни одна из следующих глав не встанет.

Одно значение и поток значений

Сравните две подписи. Обе про деньги, но ведут себя по-разному:

suspend fun getBalance(): Int        // спросил — получил число, всё
fun prices(): Flow<Int>              // подписался — числа идут, пока слушаешь

Первая — вопрос с ответом. Вторая — подписка на рассылку.

Запрос отдаёт одно значение и завершается; поток присылает значения, пока подписка жива

Обратите внимание на слово в сигнатуре: у второй функции нет suspend. Она ничего не ждёт — просто отдаёт описание будущего потока. Ждать будет тот, кто начнёт его слушать, и вот он уже обязан быть в корутине.

Первый поток: `flow` и `collect`

Самый прямой способ создать поток — построитель flow { }. Внутри него доступна функция emit — «отдать очередное значение».

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

fun main() = runBlocking {
  val numbers = flow {
      for (i in 1..3) {
          delay(100)
          println("отдаю $i")
          emit(i)
      }
  }
  println("поток описан, но ничего не произошло")
  numbers.collect { value -> println("получил $value") }
}

Живой вывод:

поток описан, но ничего не произошло
отдаю 1
получил 1
отдаю 2
получил 2
отдаю 3
получил 3

Прочитайте первые две строки ещё раз. «Поток описан, но ничего не произошло» напечаталось раньше, чем «отдаю 1» — хотя в коде println стоит после объявления numbers.

Это не опечатка и не случайность. Пока никто не начал слушать, тело flow { } не выполняется вовсе. И дальше видно, что значения идут не пачкой в конце, а по одному: отдал — получил, отдал — получил.

Нажми на часть выражения

Отвечено вопросов: 0 из 1

Почему строка «поток описан, но ничего не произошло» напечаталась раньше, чем «отдаю 1»?

Отвечено верно: 0 из 1

Холодный поток: конвейер, который стоит

То свойство, которое вы только что видели, называется холодностью. Холодный поток — это рецепт, а не готовое блюдо. Он хранит инструкцию «что делать», и выполняет её заново для каждого, кто подписался.

Операторы только описывают обработку; работу запускает единственный вызов collect

Отсюда два практических следствия, которые важнее самого термина.

Первое: два collect — две отдельные работы. Если подписаться на один и тот же поток дважды, тело flow { } выполнится дважды, и сетевой запрос внутри уйдёт два раза. Не «данные раздадутся двум подписчикам», а именно два независимых прогона.

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

Интересный факт

Слово «холодный» пришло из библиотек реактивного программирования, где противопоставление cold/hot появилось раньше Flow. Горячий поток — тот, что идёт независимо от подписчиков: значения в нём происходят, даже когда никто не слушает, и опоздавший подписчик прошлое не увидит. С таким мы встретимся в конце главы.

Операторы: `map` и `filter`

Между источником и подписчиком можно поставить обработку. Операторы выглядят как знакомые функции коллекций из главы про коллекции, и работают похоже — с одной поправкой, которая всё меняет.

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

fun main() = runBlocking {
  flowOf(1, 2, 3, 4, 5)
      .filter { it % 2 == 1 }
      .map { it * 10 }
      .collect { println(it) }
}

Вывод — 10, 30, 50.

Поправка вот в чём: у списка filter сразу создаёт новый список, его можно распечатать. У потока filter не создаёт ничего — он возвращает описание нового потока, где значения будут отфильтрованы, когда пойдут. Сравните мысленно:

listOf(1, 2, 3).filter { it > 1 }   // здесь уже есть готовый список [2, 3]
flowOf(1, 2, 3).filter { it > 1 }   // здесь есть только план: «буду пропускать те, что больше 1»

flowOf(...) — короткий способ сделать поток из готовых значений, удобен для примеров и тестов. Список тоже превращается в поток одной строкой: listOf(1, 2, 3).asFlow().

Оператор	Что делает
`map { }`	превращает каждое значение в другое
`filter { }`	пропускает дальше только подходящие
`take(n)`	берёт первые n значений и закрывает поток
`onEach { }`	подсматривает значение, не меняя его — удобно для логов
`distinctUntilChanged()`	не пропускает значение, если оно совпало с предыдущим
`debounce(ms)`	ждёт паузу — тот самый приём для поля поиска

Отвечено вопросов: 0 из 1

Чем `filter` у потока отличается от `filter` у списка?

Отвечено верно: 0 из 1

`collect` работает только в корутине

collect — suspend-функция. Это логично: она ждёт значения, а ждать умеют только корутины. Но забыть об этом легко, и компилятор отвечает так:

import kotlinx.coroutines.flow.*

fun main() {
  flowOf(1, 2).collect { println(it) }
}

Suspend function 'suspend fun collect(collector: FlowCollector<Int>): Unit'
can only be called from a coroutine or another suspend function.

Сообщение прямое: собирать поток можно только из корутины или из другой suspend-функции. Лечится тем же способом, что и в прошлой главе, — runBlocking в учебном примере, viewModelScope в настоящем приложении.

Ошибки внутри потока: `catch` и `onCompletion`

Внутри flow { } живёт настоящий код — сетевой запрос, чтение базы. Значит, он может упасть. Оборачивать collect в try/catch — рабочий, но грубый способ: он ловит и ошибки источника, и ошибки самого обработчика, не различая их.

Аккуратнее поставить оператор catch — он ловит только то, что случилось выше по конвейеру:

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

fun main() = runBlocking {
  flow {
      emit(1)
      emit(2)
      throw IllegalStateException("сервер не ответил")
  }
      .catch { e -> println("поймали: ${e.message}") ; emit(-1) }
      .onCompletion { println("поток закончился") }
      .collect { println("значение $it") }
}

Живой вывод:

значение 1
значение 2
поймали: сервер не ответил
значение -1
поток закончился

Три вещи, которые стоит заметить. Значения 1 и 2 дошли до подписчика до падения — поток отдаёт по мере готовности, а не всё в конце. Внутри catch разрешён emit — можно подставить запасное значение, и подписчик получит -1 вместо падения. А onCompletion сработал в любом случае, и сработал последним — это место для того, что должно случиться независимо от исхода, вроде выключения индикатора загрузки.

Важно

catch ловит только то, что произошло выше него в цепочке. Ошибка внутри самого collect { } через него не пройдёт — она случилась ниже. Поэтому catch ставят как можно ближе к концу конвейера, но перед collect, а в самом collect держат только отрисовку, без логики, способной упасть.

`StateFlow`: у состояния всегда есть значение

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

StateFlow решает ровно это: у него всегда есть текущее значение, доступное прямо сейчас через .value.

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

fun main() = runBlocking {
  val state = MutableStateFlow("загрузка")
  println("значение есть сразу: ${state.value}")

  val job = launch { state.collect { println("экран показывает: $it") } }
  delay(50)
  state.value = "готово"
  delay(50)
  state.value = "готово"
  delay(50)
  job.cancel()
}

Живой вывод:

значение есть сразу: загрузка
экран показывает: загрузка
экран показывает: готово

Здесь спрятан важный факт, который легко пропустить. Значение "готово" присвоено дважды, а напечаталось один раз.

StateFlow сравнивает новое значение с текущим и, если они равны, никого не будит. Внутри него встроен distinctUntilChanged. Для экрана это подарок: лишняя перерисовка не случится сама собой. Но это же и ловушка — если менять поле внутри объекта, не создавая новый, StateFlow увидит тот же самый объект и промолчит. Поэтому состояние экрана держат в data class и обновляют через copy().

Отвечено вопросов: 0 из 1

Почему во втором присваивании state.value = «готово» подписчик ничего не получил?

Отвечено верно: 0 из 1

Кто пишет и кто читает

В ViewModel встретится пара из двух свойств, и это не дублирование, а разделение прав:

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

MutableStateFlow умеет и читать, и писать — он приватный, менять состояние может только сама ViewModel. Наружу отдаётся StateFlow без слова Mutable: у него нет .value =, только чтение. Экран физически не может присвоить состояние в обход логики — не потому что «так договорились», а потому что у типа нет такого метода.

Подчёркивание в начале имени — не требование языка, а общепринятая пометка «это внутреннее, снаружи не трогать».

`SharedFlow`: событие вместо состояния

Не всё на экране — состояние. «Показать снекбар "Сохранено"» — не состояние, а происшествие. Разница не философская, а очень практическая: если положить такое в StateFlow, то после поворота экрана подписка возобновится, получит текущее значение — и снекбар покажется второй раз.

Для происшествий есть SharedFlow. У него нет .value, он ничего не хранит и отдаёт значения только тем, кто слушает в момент события.

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

fun main() = runBlocking {
  val events = MutableSharedFlow<String>()
  val job = launch { events.collect { println("подписчик увидел: $it") } }
  delay(50)
  events.emit("сохранено")
  events.emit("сохранено")
  delay(50)
  job.cancel()
  println("а StateFlow отдал бы значение сразу при подписке")
}

подписчик увидел: сохранено
подписчик увидел: сохранено
а StateFlow отдал бы значение сразу при подписке

Сравните с предыдущим примером: там два одинаковых значения дали одну строку, здесь — две. SharedFlow не сравнивает значения, потому что два одинаковых события — это два разных события. Два раза нажали «Сохранить» — два снекбара, и это правильно.

StateFlow хранит текущее состояние, SharedFlow передаёт разовые события

Как выбрать за пять секунд. Задайте вопрос: «если экран пересоздать, это должно появиться снова?»

Список товаров, флаг загрузки, текст в поле — должны. Это StateFlow. Снекбар, переход на другой экран, вибрация — не должны. Это SharedFlow.

Отвечено вопросов: 0 из 1

Куда положить событие «показать снекбар об успешном сохранении»?

Отвечено верно: 0 из 1

Где это встретится в приложении

Полную картину соберём в следующих главах, но общий вид полезно увидеть заранее — тогда будет ясно, ради чего эта глава.

// в ViewModel: состояние наружу только для чтения
private val _uiState = MutableStateFlow(UiState())
val uiState: StateFlow<UiState> = _uiState

// в Composable: подписка, которая сама встаёт на паузу вместе с экраном
val state by viewModel.uiState.collectAsStateWithLifecycle()

collectAsStateWithLifecycle — тот самый способ подписаться из Compose. Он останавливает сбор, когда экран уходит с переднего плана, и возобновляет при возврате. Обычный collectAsState этого не делает и продолжает слушать невидимый экран, тратя батарею и трафик.

Дальше по треку Flow появится ещё дважды: запрос к базе будет возвращать Flow, благодаря чему список на экране обновится сам после записи, а кэш соберётся из двух источников — потока из базы и разового запроса в сеть.

Типичные ошибки и их настоящие сообщения

Забыли, что collect — suspend.

Suspend function 'suspend fun collect(collector: FlowCollector<Int>): Unit'
can only be called from a coroutine or another suspend function.

Вызвали сбор из обычной функции. Нужна корутина: viewModelScope.launch { ... }.

Ждут значений от холодного потока, не подписавшись. Ошибки нет вообще — программа просто молчит. Это неприятнее исключения: код выглядит рабочим. Если поток «ничего не присылает», первым делом проверьте, вызван ли collect.

Меняют поле внутри объекта в StateFlow. Экран не перерисовывается, ошибок нет. Причина — StateFlow сравнил новое значение с текущим через equals и решил, что ничего не изменилось: объект-то остался тем же самым. Лечится copy(): _uiState.value = _uiState.value.copy(isLoading = false).

Кладут разовое событие в StateFlow. Снекбар появляется второй раз после поворота экрана. Причина в том, что новая подписка сразу получает хранимое значение. Событиям — SharedFlow.

Собирают поток обычным collectAsState. Сбор продолжается, когда экран ушёл с переднего плана: тратятся батарея и трафик, а данные приходят в невидимый экран. collectAsStateWithLifecycle приостанавливает сбор вместе с жизненным циклом и возобновляет при возврате.

Слова и символы главы

Flow — последовательность значений, растянутая во времени.
Холодный поток — выполняется заново для каждого подписчика; без подписки не делает ничего.
Горячий поток — существует независимо от подписчиков; опоздавший прошлое не увидит.
emit(x) — отдать очередное значение. Доступен внутри flow { }.
collect { } — начать слушать. suspend, только из корутины.
Оператор — map, filter и прочие: описывают обработку, не запуская её.
StateFlow — горячий поток с текущим значением; равное значение не рассылает.
SharedFlow — горячий поток без значения, для разовых событий.
MutableStateFlow — то же, что StateFlow, но с правом записи через .value.
collectAsStateWithLifecycle — подписка из Compose, уважающая жизненный цикл экрана.

Тренажёр: набрать вслепую

На чемпионате эти строки набираются десятки раз, и привозить заготовки нельзя. Наберите по памяти, не подглядывая.

Цель: скорость ≥ 100 зн/мин, точность ≥ 88%

private val _uiState = MutableStateFlow(UiState())

Что дальше

Flow закрыл вопрос «откуда экран берёт меняющиеся данные». Следующий вопрос — кто эти данные держит и почему они не пропадают при повороте экрана. Это глава про ViewModel: там пара MutableStateFlow и StateFlow из этой главы окажется на своём рабочем месте, а UiState перестанет быть абстрактным примером.

Отвечено вопросов: 0 из 1

Что произойдёт, если подписаться на один и тот же холодный Flow из двух разных мест?

Отвечено верно: 0 из 1

Углубиться

Операторы объединения: combine соединяет два потока и срабатывает при изменении любого — так собирают экран из состояния фильтра и списка данных.
flatMapLatest — приём для поиска: новый запрос отменяет предыдущий, и на экран не попадёт устаревший ответ.
stateIn превращает холодный поток в StateFlow с заданной политикой подписки — так репозиторий отдаёт данные, а ViewModel делает из них состояние.
flowOn меняет поток выполнения для верхней части конвейера, не трогая нижнюю.
Буферизация: buffer, conflate, collectLatest — что делать, когда значения приходят быстрее, чем обрабатываются.

Сводная проверка по всей главе: 7 вопросов, 5:00 на всё, без подсказок по ходу. Вопросы показываются по одному, ответ изменить нельзя. Если время выйдет — экзамен завершится с тем, что успел ответить. Пересдавать можно сколько угодно раз.

Куда дальше: проверенные ресурсы

Asynchronous Flow — официальный гид kotlinx.coroutines — первоисточник по построителям и операторам.
StateFlow and SharedFlow — developer.android.com — разбор пары «состояние и событие» с примерами из Android.
Kotlin flows on Android — как поток проходит через слои приложения, включая collectAsStateWithLifecycle.
Things to know about Flow's shareIn and stateIn — про превращение холодного потока в горячий.

Видео по теме

Видео#7 Kotlin Coroutines Flow. StateFlow. SharedFlow || Курс по корутинамAndroid Broadcast. Все об Андроид разработке

ВидеоKotlin Flow. SharedFlow. StateFlow [Ru, Kotlin] / Мобильный разработчикАлексей Гладков

ВидеоHot Flows vs. Cold Flows In Kotlin - When to Use What?Philipp Lackner

ВидеоМаксим Сидоров — Kotlin Flow и реальные задачи SberDevicesJPoint, Joker и JUG ru — Java-конференции

ВидеоМиграция с LiveData на Coroutines FlowAndroid Broadcast. Все об Андроид разработке

Предложить правку

Зачем это нужно​

Одно значение и поток значений​

Первый поток: flow и collect​

Холодный поток: конвейер, который стоит​

Операторы: map и filter​

collect работает только в корутине​

Ошибки внутри потока: catch и onCompletion​

StateFlow: у состояния всегда есть значение​

Кто пишет и кто читает​

SharedFlow: событие вместо состояния​

Где это встретится в приложении​

Типичные ошибки и их настоящие сообщения​

Слова и символы главы​

Тренажёр: набрать вслепую​

Что дальше​

Углубиться​

Куда дальше: проверенные ресурсы​

Видео по теме

Комментарии

Зачем это нужно

Одно значение и поток значений

Первый поток: `flow` и `collect`

Холодный поток: конвейер, который стоит

Операторы: `map` и `filter`

`collect` работает только в корутине

Ошибки внутри потока: `catch` и `onCompletion`

`StateFlow`: у состояния всегда есть значение

Кто пишет и кто читает

`SharedFlow`: событие вместо состояния

Где это встретится в приложении

Типичные ошибки и их настоящие сообщения

Слова и символы главы

Тренажёр: набрать вслепую

Что дальше

Углубиться

Куда дальше: проверенные ресурсы