当前位置：首页 > article >正文

Kotlin 流 Flow

article 2024/11/14 5:50:47

挂起函数可以异步地返回一个值，而对于返回多个值，可以使用流，使用 emit(x) 发射多个值，
collect { } 来收集值。

默认流是冷的，只有 收集（collect） 时才会执行。

1. 流的创建

flow {} 生成流，emit(x) 来发射值；
xxx.asFlow() 集合转成Flow；
flowOf(1, 2, 3) 生成固定值的流。

1.1 flow {}

flow {} 里的发射（emissions）默认是可取消的，对应 SafeFlow，继承自 AbstractFlow：

public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {

    public final override suspend fun collect(collector: FlowCollector<T>) {
        val safeCollector = SafeCollector(collector, coroutineContext)
        try {
            collectSafely(safeCollector)
        } finally {
            safeCollector.releaseIntercepted()
        }
    }

}

对应 emit() 方法，就是 SafeCollector.emit() 内部调用了 currentContext.ensureActive() 做取消检查。

而其他两种，默认是 不可取消，使用 cancellable() 操作符可取消。

2. 流的操作符

2.1 中间操作符

map
filter
transform

2.2 结束操作符

collect {}
toList() 或 toSet() 收集到集合
first() 或 single()
reduce() 或 fold() 合并值

3. 流的上下文

默认流执行在和 收集者（Collector） 相同的上下文。

更改流的发射的上下文，必须使用 flowOn，而不是在 flow {} 中使用 withContext()。

flow {
    ...
}.flowOn(Dispatchers.Default)

4. 缓冲

当 Flow emit 生产者速度大于 collector 消费者速度时。

buffer() 并发地执行发射和收集，而不是顺序执行（发射收集再发射收集）；
conflate() 丢弃中间值，取最新发射的值；
collectLatest { } 收集最新的值，但如果发射新值，会取消慢的收集。

simple()
    .collectLatest { value -> // cancel & restart on the latest value
        println("Collecting $value") 
        delay(300) // pretend we are processing it for 300 ms
        println("Done $value") 
    }

说明：会执行所有 Collecting，但是因为慢处理，会被取消。

结果：

Collecting 1
Collecting 2
Collecting 3
Done 3

https://kotlinlang.org/docs/flow.html#processing-the-latest-value

5. 流的组合

zip 1对1的组合
combine 每次上游更新，就会重新计算。两个流生产速度不一样时，就会不同的对应组合。更新值即组合

举例：

zip 组合值: 1->one, 2->two 3->three，
而 combine 组合，则可能 1->one 2->one 这样只有一方流发射值，就会调用计算。

6. flatten 展平

对于 Flow 又对应 Flow<T> 任务，这时候对于 Flow<Flow<T>> 需要展开为Flow<T>。场景就是一序列对应请求任务。

fun requestFlow(i: Int): Flow<String> = flow {
    emit("$i: First")
    delay(500) // wait 500 ms
    emit("$i: Second")
}

(1..3).asFlow().map { requestFlow(it) }

flatMapConcat 按顺序，流完成后，才接着下一个流
flatMapMerge 支持并发地处理，流则出现并发交错地收集值，concurrency 设置并发数
flatMapLatest 处理最新的流，当新的流发射值时，取消之前的流

7. 异常

try { flow.collect {} } catch (e: Exception) { } 处理异常，包含收集器里代码异常；
flow {}.catch { }.collect { } 处理上游异常，但不会处理下游异常；
flow {}.onEach { }.collect() 处理上下游异常。

7.1. `try/catch` 全部捕获

try {
    simple()
        .collect {
            println("value: $it")
        }
} catch (e: Exception) {
    // 捕获了 flow发射代码块、中间操作符 和 结束操作符 的所有异常
}

7.2. `catch` 操作符

simple()
    .catch { e ->
        // catch 捕获上面的异常，但 不处理 下游 和 结束操作符 的 异常
        println("exception: $e") 
    }
    .collect {
        // 如果 这里异常，则不会被捕捉
        println("value: $it")
    }

7.3. 声明式捕获

如果想捕获结束操作符的异常，需要声明式地捕捉。把 collect 的代码部分上移到 onEach 中，使用无参的 collect() 收集：

simple()
    .onEach {
        check(it < 2)
        println("value: $it")
    }
    .catch { e -> 
        println("exception: $e")
    }
    .collect()

8. 完成

try/finnaly 在结束后处理
flow {}.onCompletion { cause -> } 处理

flow {

}.onCompletion { cause ->
    // 完成回调，cause 是空 表示 正常完成
    if (cause == null) {
        println("success")
    }
}

9. 取消

onEach 时检测
cancel() 在收集时，调用取消
flow {}.cancellable() 设置flow可取消

// 不加 cancellable() 不会 做取消检查，导致完成收集后 才 报异常
// cancellable() 则会 及时取消
// flowOf(1, 2, 3)
listOf(1, 2, 3).asFlow()
//  .cancellable()
    .collect {
        if (it > 1) {
            cancel()
        }
        println("value: $it")
    }

cancellable()的实现：

CancellableFlowImpl

public fun <T> Flow<T>.cancellable(): Flow<T> =
    when (this) {
        is CancellableFlow<*> -> this // Fast-path, already cancellable
        else -> CancellableFlowImpl(this)
    }

private class CancellableFlowImpl<T>(private val flow: Flow<T>) : CancellableFlow<T> {
    override suspend fun collect(collector: FlowCollector<T>) {
        flow.collect {
            currentCoroutineContext().ensureActive()
            collector.emit(it)
        }
    }
}