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

Flow原理

news 来源：原创 2025/4/29 5:39:27

fun main() {runBlocking {launch {flow4.collect{println("---collect-4")}println("---flow4")}}val flow4 = flow<Boolean>{delay(5000)emit(false)
}

我们分析下整个流程

1.flow为什么之后在collect之后才会发送数据

2.collect的调用流程

我们先看创建flow的方法

public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)

可以看见是一个方法，返回的是一个SafeFlow对象，然后把我们传入 {
delay(5000)
emit(false)
} 表达式的传入到了这个对象中。

private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {override suspend fun collectSafely(collector: FlowCollector<T>) {collector.block()}
}

然后，这个对象提供一个方法collectSafely，这个时候可以看到，要发送数据，必须得调用SafeFlow的collectSafely方法才行。

接下来我们分析下collect方法。看源码发现需要传入一个FlowCollector接口实现类

public suspend fun collect(collector: FlowCollector<T>)

public fun interface FlowCollector<in T> {//注意这个emit方法public suspend fun emit(value: T)
}

因为我们是使用flow方法返回的SafeFlow对象去调用的，所以我们看下SafeFlow的collect方法。SafeFlow是继承AbstractFlow的类，所以我们看这个类就行

public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {public final override suspend fun collect(collector: FlowCollector<T>) {//创建SafeCollector对象val safeCollector = SafeCollector(collector, coroutineContext)try {//调用实现类的collectSafely方法，把SafeCollector对象传递过去collectSafely(safeCollector)} finally {safeCollector.releaseIntercepted()}}public abstract suspend fun collectSafely(collector: FlowCollector<T>)
}

再接着查看collectSafely方法,发现调用到了我们传入的闭包

private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {override suspend fun collectSafely(collector: FlowCollector<T>) {//调用我们传入的闭包，而且他是FlowCollector的扩展函数collector.block()}
}

而我们传入的闭包是suspend FlowCollector<T>.() -> Unit扩展函数，这个对象就是我们collect传入的接口实现类，所以在

val flow4 = flow<Boolean>{
delay(5000)
emit(false)
}

方法中调用emit（）实际就是调用collect传入接口实现类的emit方法