784 lines
33 KiB
Markdown
784 lines
33 KiB
Markdown
---
|
||
title: Flow
|
||
description: Холодный асинхронный поток значений: построение, операторы, контекст, обработка ошибок и backpressure
|
||
tags:
|
||
- kotlin
|
||
- coroutines
|
||
- flow
|
||
---
|
||
|
||
`Flow<T>` - асинхронная последовательность значений. Если `suspend`-функция возвращает одно значение, то `Flow` - много, причём каждое может приходить с задержкой.
|
||
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun temperatures(): Flow<Int> = flow {
|
||
for (t in listOf(20, 22, 25)) {
|
||
delay(20) // внутри flow можно приостанавливаться
|
||
emit(t)
|
||
}
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
temperatures().collect { println("температура: $it") }
|
||
// температура: 20
|
||
// температура: 22
|
||
// температура: 25
|
||
}
|
||
```
|
||
Сравните с `List`: список нужно вычислить целиком, прежде чем вернуть. `Flow` отдаёт значения по мере готовности.
|
||
|
||
## Холодный поток
|
||
`Flow` ничего не делает, пока его не начнут собирать. Каждый вызов `collect` запускает тело заново.
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
var starts = 0
|
||
val f = flow {
|
||
starts++
|
||
emit(1)
|
||
emit(2)
|
||
}
|
||
|
||
println("создан, запусков: $starts") // создан, запусков: 0
|
||
f.collect { }
|
||
f.collect { }
|
||
println("после двух collect: $starts") // после двух collect: 2
|
||
}
|
||
```
|
||
Это главное свойство: `Flow` - **описание** вычисления, а не запущенный процесс. Горячие потоки, которые работают независимо от подписчиков, - это `StateFlow` и `SharedFlow`. Подробнее: [StateFlow и SharedFlow](StateFlow%20и%20SharedFlow.md)
|
||
|
||
## Способы создать Flow
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2, 3).collect { print("$it ") }
|
||
println() // 1 2 3
|
||
|
||
listOf("a", "b").asFlow().collect { print("$it ") }
|
||
println() // a b
|
||
|
||
(1..3).asFlow().collect { print("$it ") }
|
||
println() // 1 2 3
|
||
|
||
flow {
|
||
emit("из билдера")
|
||
emitAll(flowOf("и", "вложенного"))
|
||
}.collect { print("$it ") }
|
||
println() // из билдера и вложенного
|
||
}
|
||
```
|
||
Для колбэчных источников есть `callbackFlow` - он же мост от слушателей к `Flow`:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.channels.awaitClose
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
class Sensor {
|
||
private var listener: ((Int) -> Unit)? = null
|
||
fun subscribe(l: (Int) -> Unit) { listener = l }
|
||
fun unsubscribe() { listener = null; println("слушатель снят") }
|
||
fun fire(value: Int) = listener?.invoke(value)
|
||
}
|
||
|
||
fun Sensor.readings(): Flow<Int> = callbackFlow {
|
||
subscribe { value -> trySend(value) }
|
||
awaitClose { unsubscribe() } // вызовется при отмене collect
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
val sensor = Sensor()
|
||
val job = launch {
|
||
sensor.readings().collect { println("показание: $it") }
|
||
}
|
||
delay(20)
|
||
sensor.fire(42)
|
||
sensor.fire(43)
|
||
delay(20)
|
||
job.cancelAndJoin()
|
||
// показание: 42
|
||
// показание: 43
|
||
// слушатель снят
|
||
}
|
||
```
|
||
`awaitClose` обязателен: без него `callbackFlow` бросит `IllegalStateException`, потому что иначе слушатель остался бы висеть.
|
||
|
||
## Промежуточные операторы
|
||
Операторы ленивые: они только оборачивают поток, не запуская его.
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val result = (1..10).asFlow()
|
||
.filter { it % 2 == 0 }
|
||
.map { it * it }
|
||
.take(3)
|
||
.toList()
|
||
println(result) // [4, 16, 36]
|
||
}
|
||
```
|
||
Главное отличие от операторов коллекций - внутри можно приостанавливаться:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
suspend fun enrich(id: Int): String { delay(10); return "запись $id" }
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2, 3)
|
||
.map { enrich(it) } // suspend внутри map
|
||
.collect { println(it) }
|
||
// запись 1
|
||
// запись 2
|
||
// запись 3
|
||
}
|
||
```
|
||
|
||
| Оператор | Что делает |
|
||
|---|---|
|
||
| `map`, `filter` | как у коллекций, но с `suspend` внутри |
|
||
| `transform` | произвольное число `emit` на элемент |
|
||
| `take`, `drop`, `takeWhile` | ограничение по количеству |
|
||
| `onEach` | побочный эффект без изменения значений |
|
||
| `withIndex`, `runningFold`, `scan` | накопление |
|
||
| `distinctUntilChanged` | пропуск подряд идущих дублей |
|
||
| `debounce`, `sample` | фильтрация по времени |
|
||
| `flatMapConcat`, `flatMapMerge`, `flatMapLatest` | поток из потоков |
|
||
| `zip`, `combine` | объединение двух потоков |
|
||
| `onStart`, `onCompletion`, `catch` | обработка жизненного цикла |
|
||
|
||
`transform` - самый общий:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2, 3)
|
||
.transform { value ->
|
||
emit("начало $value")
|
||
emit("конец $value")
|
||
}
|
||
.collect { print("$it; ") }
|
||
println()
|
||
// начало 1; конец 1; начало 2; конец 2; начало 3; конец 3;
|
||
}
|
||
```
|
||
|
||
## Терминальные операторы
|
||
Только они запускают поток, и все они `suspend`:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val f = flowOf(1, 2, 3, 4)
|
||
|
||
println(f.toList()) // [1, 2, 3, 4]
|
||
println(f.first()) // 1
|
||
println(f.firstOrNull { it > 10 }) // null
|
||
println(f.lastOrNull()) // 4
|
||
println(f.count()) // 4
|
||
println(f.reduce { a, b -> a + b }) // 10
|
||
println(f.fold(100) { a, b -> a + b }) // 110
|
||
println(f.toSet()) // [1, 2, 3, 4]
|
||
println(f.filter { it == 2 }.single()) // 2
|
||
}
|
||
```
|
||
`first()` отменяет поток сразу после первого значения - этого достаточно, чтобы остановить бесконечный источник:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
var emitted = 0
|
||
val infinite = flow {
|
||
var i = 0
|
||
while (true) {
|
||
emitted++
|
||
emit(i++)
|
||
}
|
||
}
|
||
println(infinite.first()) // 0
|
||
println("излучено значений: $emitted") // излучено значений: 1
|
||
}
|
||
```
|
||
|
||
## Контекст и flowOn
|
||
По умолчанию `Flow` выполняется в контексте того, кто его собирает - это называется context preservation:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow {
|
||
println("flow на: ${Thread.currentThread().name}")
|
||
emit(1)
|
||
}.collect {
|
||
println("collect на: ${Thread.currentThread().name}")
|
||
}
|
||
// flow на: main
|
||
// collect на: main
|
||
}
|
||
```
|
||
Сменить контекст **производителя** можно только через `flowOn`:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow {
|
||
val onPool = Thread.currentThread().name.contains("DefaultDispatcher")
|
||
println("flow в пуле: $onPool")
|
||
emit(1)
|
||
}
|
||
.flowOn(Dispatchers.IO)
|
||
.collect {
|
||
println("collect на: ${Thread.currentThread().name}")
|
||
}
|
||
// flow в пуле: true
|
||
// collect на: main
|
||
}
|
||
```
|
||
`flowOn` действует на всё, что **выше** него по цепочке:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2)
|
||
.map { it * 10 } // в IO
|
||
.flowOn(Dispatchers.IO)
|
||
.map { it + 1 } // в main
|
||
.collect { print("$it ") }
|
||
println() // 11 21
|
||
}
|
||
```
|
||
Менять контекст внутри `flow { }` через `withContext` запрещено:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val broken = flow {
|
||
withContext(Dispatchers.IO) { emit(1) } // нарушение контракта
|
||
}
|
||
val error = runCatching { broken.collect { } }.exceptionOrNull()
|
||
println(error is IllegalStateException) // true
|
||
println(error?.message?.contains("Flow invariant is violated")) // true
|
||
}
|
||
```
|
||
Причина запрета - предсказуемость: собирающий код должен быть уверен, что `collect` вызывается в его контексте (например, в UI-потоке). Для смены контекста есть только `flowOn`.
|
||
|
||
## Обработка ошибок
|
||
`try/catch` работает, но правильнее оператор `catch`:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow {
|
||
emit(1)
|
||
throw IllegalStateException("источник упал")
|
||
}
|
||
.catch { e -> println("перехвачено: ${e.message}") }
|
||
.collect { println("значение: $it") }
|
||
// значение: 1
|
||
// перехвачено: источник упал
|
||
}
|
||
```
|
||
`catch` ловит ошибки только **выше** себя - это называется прозрачность исключений:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val error = runCatching {
|
||
flowOf(1)
|
||
.catch { println("не сработает") }
|
||
.collect { throw IllegalStateException("ошибка в collect") }
|
||
}.exceptionOrNull()
|
||
println("ошибка collect прошла мимо catch: ${error?.message}")
|
||
// ошибка collect прошла мимо catch: ошибка в collect
|
||
}
|
||
```
|
||
Чтобы ловить и ошибки потребителя, нужен `onEach` + `catch` в таком порядке:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2)
|
||
.onEach { if (it == 2) error("обработка упала на $it") }
|
||
.catch { e -> println("перехвачено: ${e.message}") }
|
||
.collect { println("обработано: $it") }
|
||
// обработано: 1
|
||
// перехвачено: обработка упала на 2
|
||
}
|
||
```
|
||
`catch` может подменить поток значениями по умолчанию:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow<String> { error("сеть недоступна") }
|
||
.catch { emit("данные из кеша") }
|
||
.collect { println(it) }
|
||
// данные из кеша
|
||
}
|
||
```
|
||
|
||
## retry и повторные попытки
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
var attempt = 0
|
||
flow {
|
||
attempt++
|
||
if (attempt < 3) error("попытка $attempt не удалась")
|
||
emit("успех на попытке $attempt")
|
||
}
|
||
.retry(3) { e ->
|
||
println("ретрай после: ${e.message}")
|
||
delay(10)
|
||
true
|
||
}
|
||
.collect { println(it) }
|
||
// ретрай после: попытка 1 не удалась
|
||
// ретрай после: попытка 2 не удалась
|
||
// успех на попытке 3
|
||
}
|
||
```
|
||
`retryWhen` даёт доступ к номеру попытки - это нужно для экспоненциальной задержки:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
var calls = 0
|
||
flow {
|
||
calls++
|
||
if (calls < 3) error("сбой")
|
||
emit("готово")
|
||
}
|
||
.retryWhen { _, attempt ->
|
||
if (attempt >= 3) false
|
||
else {
|
||
delay(10L * (attempt + 1)) // 10, 20, 30 мс
|
||
println("попытка ${attempt + 1}")
|
||
true
|
||
}
|
||
}
|
||
.collect { println(it) }
|
||
// попытка 1
|
||
// попытка 2
|
||
// готово
|
||
}
|
||
```
|
||
|
||
## Жизненный цикл: onStart, onCompletion
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flowOf(1, 2)
|
||
.onStart { println("начали") }
|
||
.onEach { println("значение $it") }
|
||
.onCompletion { cause -> println("завершили, ошибка: $cause") }
|
||
.collect { }
|
||
// начали
|
||
// значение 1
|
||
// значение 2
|
||
// завершили, ошибка: null
|
||
}
|
||
```
|
||
`onCompletion` вызывается всегда - при успехе, ошибке и отмене, причём получает причину:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow {
|
||
emit(1)
|
||
error("упало")
|
||
}
|
||
.onCompletion { cause -> println("причина: ${cause?.message}") }
|
||
.catch { }
|
||
.collect { }
|
||
// причина: упало
|
||
}
|
||
```
|
||
`onStart` может излучать значения - это способ показать «загрузка...» до первого реального элемента:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
flow { delay(20); emit("данные") }
|
||
.onStart { emit("загрузка...") }
|
||
.collect { println(it) }
|
||
// загрузка...
|
||
// данные
|
||
}
|
||
```
|
||
|
||
## buffer: производитель и потребитель параллельно
|
||
По умолчанию производитель ждёт потребителя:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
import kotlin.system.measureTimeMillis
|
||
|
||
fun source() = flow {
|
||
repeat(3) {
|
||
delay(20) // производство
|
||
emit(it)
|
||
}
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
val sequential = measureTimeMillis {
|
||
source().collect { delay(20) } // потребление
|
||
}
|
||
val buffered = measureTimeMillis {
|
||
source().buffer().collect { delay(20) }
|
||
}
|
||
println("без буфера около 120 мс: ${sequential >= 110}")
|
||
println("с буфером быстрее: ${buffered < sequential}")
|
||
// без буфера около 120 мс: true
|
||
// с буфером быстрее: true
|
||
}
|
||
```
|
||
Без буфера шаги складываются: 3 × (20 + 20) = 120 мс. С буфером производство идёт параллельно с потреблением, и общее время приближается к максимуму из двух конвейеров.
|
||
|
||
## conflate и collectLatest
|
||
Когда важны только свежие значения, промежуточные можно отбрасывать:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun fast() = flow {
|
||
repeat(5) {
|
||
delay(10)
|
||
emit(it)
|
||
}
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
val all = mutableListOf<Int>()
|
||
fast().collect { delay(40); all += it }
|
||
|
||
val conflated = mutableListOf<Int>()
|
||
fast().conflate().collect { delay(40); conflated += it }
|
||
|
||
println("без conflate получено всё: ${all.size == 5}")
|
||
println("с conflate пропущены промежуточные: ${conflated.size < 5}")
|
||
println("последнее значение всегда доходит: ${conflated.last() == 4}")
|
||
// без conflate получено всё: true
|
||
// с conflate пропущены промежуточные: true
|
||
// последнее значение всегда доходит: true
|
||
}
|
||
```
|
||
`collectLatest` работает иначе: он не пропускает значения, а **отменяет** обработку предыдущего:
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val completed = mutableListOf<Int>()
|
||
flow {
|
||
repeat(3) { delay(10); emit(it) }
|
||
}.collectLatest { value ->
|
||
delay(25) // не успеет для первых двух
|
||
completed += value
|
||
}
|
||
println("доработал только последний: $completed") // доработал только последний: [2]
|
||
}
|
||
```
|
||
|
||
| Стратегия | Поведение при быстром производителе |
|
||
|---|---|
|
||
| по умолчанию | производитель ждёт потребителя |
|
||
| `buffer(n)` | очередь на `n` элементов, затем ожидание |
|
||
| `buffer(onBufferOverflow = DROP_OLDEST)` | выбрасывает старые |
|
||
| `conflate()` | хранит только последнее непрочитанное |
|
||
| `collectLatest` | отменяет обработку устаревшего значения |
|
||
| `debounce(ms)` | пропускает только после паузы |
|
||
| `sample(ms)` | берёт последнее раз в интервал |
|
||
|
||
## Объединение потоков
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val letters = flowOf("a", "b", "c")
|
||
val numbers = flowOf(1, 2, 3)
|
||
|
||
println(letters.zip(numbers) { l, n -> "$l$n" }.toList()) // [a1, b2, c3]
|
||
|
||
val slow = flow { emit(1); delay(30); emit(2) }
|
||
val fast2 = flow { emit("x"); delay(10); emit("y") }
|
||
println(slow.combine(fast2) { a, b -> "$a$b" }.toList())
|
||
// [1x, 1y, 2y]
|
||
}
|
||
```
|
||
`zip` ждёт пару - элементов будет столько, сколько в коротком потоке. `combine` излучает при каждом изменении любого из потоков, используя последние известные значения другого. Для UI (например, «форма валидна, если валидны все поля») нужен `combine`.
|
||
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
fun main() = runBlocking {
|
||
val concatenated = flowOf(1, 2).flatMapConcat { n ->
|
||
flow { emit("$n-a"); delay(10); emit("$n-b") }
|
||
}.toList()
|
||
println(concatenated) // [1-a, 1-b, 2-a, 2-b]
|
||
|
||
val latest = flowOf(1, 2, 3).flatMapLatest { n ->
|
||
flow { delay(20); emit("результат $n") }
|
||
}.toList()
|
||
println(latest) // [результат 3]
|
||
}
|
||
```
|
||
`flatMapLatest` - основной оператор для поиска по вводу: новый запрос отменяет предыдущий.
|
||
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
suspend fun search(query: String): List<String> {
|
||
delay(20)
|
||
return listOf("результат для «$query»")
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
val input = flow {
|
||
emit("к"); delay(5)
|
||
emit("ко"); delay(5)
|
||
emit("кот"); delay(100) // пользователь остановился
|
||
emit("кота")
|
||
}
|
||
|
||
input
|
||
.debounce(30) // ждём паузу в наборе
|
||
.filter { it.length >= 2 }
|
||
.flatMapLatest { search(it).asFlow() }
|
||
.collect { println(it) }
|
||
// результат для «кот»
|
||
// результат для «кота»
|
||
}
|
||
```
|
||
Это каноничный конвейер: `debounce` отсекает промежуточные нажатия, `flatMapLatest` отменяет устаревший запрос.
|
||
|
||
```mermaid
|
||
graph LR
|
||
A["ввод: к, ко, кот"] --> B["debounce(30)"]
|
||
B --> C["только 'кот'<br/>после паузы"]
|
||
C --> D["flatMapLatest"]
|
||
D --> E["запрос к API"]
|
||
E --> F["collect: результаты"]
|
||
D -.->|"пришёл новый ввод"| G["отмена<br/>предыдущего запроса"]
|
||
```
|
||
|
||
## Flow как возвращаемый тип
|
||
```kotlin
|
||
import kotlinx.coroutines.*
|
||
import kotlinx.coroutines.flow.*
|
||
|
||
class OrderRepository {
|
||
// не suspend: Flow холодный, вызов ничего не запускает
|
||
fun observeOrders(): Flow<String> = flow {
|
||
repeat(3) { i ->
|
||
delay(10)
|
||
emit("заказ $i")
|
||
}
|
||
}.flowOn(Dispatchers.IO) // производитель всегда в IO
|
||
}
|
||
|
||
fun main() = runBlocking {
|
||
OrderRepository().observeOrders().collect { println(it) }
|
||
// заказ 0
|
||
// заказ 1
|
||
// заказ 2
|
||
}
|
||
```
|
||
Правила проектирования: функция, возвращающая `Flow`, не должна быть `suspend` (создание потока мгновенно), а `flowOn` ставится внутри - вызывающий не обязан знать, где работает источник.
|
||
|
||
> [!info] Отличия от Java
|
||
> | Java | Kotlin |
|
||
> |---|---|
|
||
> | `Stream<T>` (синхронный, одноразовый) | `Sequence<T>` |
|
||
> | `Flow` из Reactive Streams | `Flow<T>` |
|
||
> | RxJava `Observable`/`Flowable` | `Flow<T>` |
|
||
> | Project Reactor `Flux<T>` | `Flow<T>` |
|
||
> | `Subscriber.request(n)` | приостановка `emit` |
|
||
> | `subscribeOn`/`observeOn` | `flowOn` |
|
||
> | `onErrorResume` | `catch` |
|
||
> | `BlockingQueue` + поток | `Channel` |
|
||
>
|
||
> **`java.util.stream.Stream` - это не аналог.** `Stream` синхронный: `stream.map(f)` не может ждать ответа сети, а терминальная операция блокирует поток до конца обработки. Прямой аналог `Stream` в Kotlin - `Sequence`. `Flow` отличается тем, что внутри каждого оператора разрешена приостановка. Подробнее: [Sequence](../Коллекции/Sequence.md)
|
||
>
|
||
> **Reactive Streams решают ту же задачу, но сложнее.** RxJava и Reactor построены на коллбэках (`onNext`, `onError`, `onComplete`) и требуют явного управления давлением через `request(n)`. В `Flow` backpressure получается бесплатно: `emit` - это `suspend`-функция, поэтому если потребитель не готов, производитель просто приостанавливается:
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> var produced = 0
|
||
> flow {
|
||
> repeat(100) { produced++; emit(it) }
|
||
> }.take(3).collect { }
|
||
> println("произведено ровно столько, сколько взяли: $produced")
|
||
> // произведено ровно столько, сколько взяли: 3
|
||
> }
|
||
> ```
|
||
> Никакого `request(n)` - producer сам останавливается на `emit`.
|
||
>
|
||
> **Стектрейсы и отладка.** В RxJava стектрейс ошибки состоит из кадров библиотеки, а место в бизнес-логике приходится искать по `onAssembly`-хукам. `Flow` - обычные `suspend`-функции, поэтому исключение приходит с кадрами вашего кода, а `try/catch`, `for` и `return` работают внутри операторов естественно.
|
||
>
|
||
> **Количество операторов.** В RxJava их несколько сотен (`flatMapIterable`, `concatMapDelayError`, `switchIfEmpty`...), потому что каждый сценарий требует отдельного оператора - внутри цепочки нельзя писать обычный код. В `Flow` операторов несколько десятков: остальное выражается через `transform` и обычные конструкции языка.
|
||
>
|
||
> **Отмена подписки.** В Rx нужно хранить `Disposable` и не забыть вызвать `dispose()` - классический источник утечек, для борьбы с которым придумали `CompositeDisposable`. `Flow` собирается внутри корутины, поэтому отмена области автоматически прекращает сбор:
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val job = launch {
|
||
> flow {
|
||
> try {
|
||
> repeat(100) { delay(10); emit(it) }
|
||
> } finally { println("источник закрыт") }
|
||
> }.collect { }
|
||
> }
|
||
> delay(30)
|
||
> job.cancelAndJoin()
|
||
> // источник закрыт
|
||
> }
|
||
> ```
|
||
>
|
||
> **Context preservation - то, чего нет в Rx.** В RxJava `observeOn` можно поставить в любом месте, и понять, на каком потоке выполнится `subscribe`, можно только прочитав всю цепочку. `Flow` гарантирует, что `collect` всегда выполняется в контексте собирающего, а `withContext` внутри `flow { }` запрещён на уровне рантайма.
|
||
>
|
||
> Сравнить: [Многопоточность в Java](../../Программирование%20на%20Java/Многопоточность/Взаимодейтсвие%20между%20потоками.md), [Функции в Java](../../Программирование%20на%20Java/Синтаксис%20и%20типы/Работа%20с%20данными%20и%20типами/Функции.md)
|
||
|
||
> [!warning] Подводные камни
|
||
> **`Flow` не запустится без терминального оператора:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> var side = 0
|
||
> flowOf(1, 2, 3).onEach { side++ } // collect не вызван
|
||
> delay(20)
|
||
> println("побочных эффектов: $side") // побочных эффектов: 0
|
||
> }
|
||
> ```
|
||
> Компилятор предупреждает о неиспользованном результате, но не всегда - например, если значение присвоено переменной.
|
||
>
|
||
> **`withContext` внутри `flow { }` бросает исключение в рантайме:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val error = runCatching {
|
||
> flow { withContext(Dispatchers.IO) { emit(1) } }.collect { }
|
||
> }.exceptionOrNull()
|
||
> println(error?.message?.contains("Flow invariant is violated")) // true
|
||
> }
|
||
> ```
|
||
> Это ошибка времени выполнения, а не компиляции - её легко пропустить в редкой ветке кода. Используйте `flowOn`.
|
||
>
|
||
> **`flowOn` действует только выше по цепочке.** `flow.map { a }.flowOn(IO).map { b }` выполнит `a` в IO, а `b` - в контексте сборщика. Порядок операторов меняет смысл.
|
||
>
|
||
> **`catch` не ловит ошибки `collect`:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val error = runCatching {
|
||
> flowOf(1).catch { println("не вызовется") }.collect { error("из collect") }
|
||
> }.exceptionOrNull()
|
||
> println("ошибка прошла мимо: ${error?.message}") // ошибка прошла мимо: из collect
|
||
> }
|
||
> ```
|
||
> Перенесите логику в `onEach` перед `catch`.
|
||
>
|
||
> **`emit` из другой корутины запрещён:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val broken = flow {
|
||
> coroutineScope {
|
||
> launch { emit(1) } // emit в другой корутине
|
||
> }
|
||
> }
|
||
> val error = runCatching { broken.collect { } }.exceptionOrNull()
|
||
> println(error != null) // true
|
||
> }
|
||
> ```
|
||
> Сообщение - про нарушение инварианта. Для излучения из нескольких источников нужен `channelFlow` или `callbackFlow`.
|
||
>
|
||
> **`callbackFlow` без `awaitClose` падает:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val broken = callbackFlow<Int> {
|
||
> trySend(1)
|
||
> // нет awaitClose
|
||
> }
|
||
> val error = runCatching { broken.collect { } }.exceptionOrNull()
|
||
> println(error is IllegalStateException) // true
|
||
> }
|
||
> ```
|
||
> Это защита от утечки слушателя: без `awaitClose` библиотека не знает, когда отписываться.
|
||
>
|
||
> **`trySend` молча теряет значения при полном буфере.** У `callbackFlow` буфер по умолчанию 64 элемента; `trySend` возвращает `ChannelResult`, который обычно игнорируют. Если источник быстрее потребителя, часть событий исчезнет - проверяйте результат или задавайте `buffer(Channel.UNLIMITED)` осознанно.
|
||
>
|
||
> **Повторный `collect` запускает поток заново.** Если `Flow` делает сетевой запрос, два `collect` - это два запроса. Для разделения результата между подписчиками нужен `shareIn`/`stateIn`. Подробнее: [StateFlow и SharedFlow](StateFlow%20и%20SharedFlow.md)
|
||
>
|
||
> **`Flow` без буфера последователен, а не параллелен:**
|
||
> ```kotlin
|
||
> import kotlinx.coroutines.*
|
||
> import kotlinx.coroutines.flow.*
|
||
> import kotlin.system.measureTimeMillis
|
||
>
|
||
> fun main() = runBlocking {
|
||
> val t = measureTimeMillis {
|
||
> flowOf(1, 2, 3).map { delay(30); it }.collect { }
|
||
> }
|
||
> println("последовательно, около 90 мс: ${t >= 85}") // последовательно, около 90 мс: true
|
||
> }
|
||
> ```
|
||
> `map` с `suspend` внутри обрабатывает элементы по одному. Для параллельной обработки нужен `flatMapMerge` с `concurrency` или `map { async { } }` поверх списка.
|
||
>
|
||
> **`conflate` и `collectLatest` теряют данные - это их цель.** Для UI-состояния это правильно, для обработки платежей или записи в журнал - нет. Убедитесь, что потеря промежуточных значений допустима.
|
||
>
|
||
> **`debounce` задерживает даже последнее значение.** Пользователь перестал печатать - результат появится только через указанный интервал. Слишком большое значение ощущается как лаг интерфейса.
|
||
>
|
||
> **Бесконечный `Flow` в `toList()` повиснет.** Терминальные операторы, собирающие всё (`toList`, `count`, `fold`), на бесконечном источнике не завершатся никогда. Нужен `take(n)` или `first()`.
|
||
>
|
||
> **`SharingStarted.Eagerly` в `stateIn` держит источник активным всегда** - типичная утечка при переносе холодного `Flow` в горячий. Подробнее: [StateFlow и SharedFlow](StateFlow%20и%20SharedFlow.md)
|
||
>
|
||
> **Исключение в `onCompletion` не перехватывается предыдущим `catch`.** `onCompletion` - не `finally`: он не может подавить ошибку, а своя ошибка внутри него уйдёт потребителю.
|
||
>
|
||
> **`first()` отменяет источник.** Для `callbackFlow` это значит, что слушатель будет снят сразу после первого значения - если вы ожидали продолжения подписки, её нет.
|
||
|
||
Далее: [StateFlow и SharedFlow](StateFlow%20и%20SharedFlow.md)
|