1.Flow原理探究
我们还是以最简单的测试代码入手:
private suspend fun testFlow() {
//上游操作符,创建Flow的同时,发射数据
flow {
emit(1)
emit(2)
emit(3)
emit(4)
emit(5)
}
//终止操作符,接收数据
.collect {
println(it)
}
}
直接看一下flow{}高阶函数:
public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)
分析如下:
flow{}的返回值是Flow类型,其实是返回一个SafeFlow对象。
block的参数是suspend FlowCollector<T>.() -> Unit,可知这段lambda是挂起函数,最终会被编译成SuspendLambda类型对象;其次它的接收者类型是FlowCollector<T>,也就是可以把block看成FlowCollector的成员方法。
这里的SafeFlow定义:
private class SafeFlow<T>(private val block: suspend FlowCollector<T>.() -> Unit) : AbstractFlow<T>() {
override suspend fun collectSafely(collector: FlowCollector<T>) {
collector.block()
}
}
SafeFlow继承至AbstractFlow,即Flow的抽象类(可以看成是Flow的基础抽象实现),重写了collectSafely方法,注意该方法中,会调用collector.block(),而这个动作就会触发lambda代码块的执行。
所以说这是一个重点方法,我们看看什么地方会调用该方法,分析其父类AbstractFlow:
//代码1
//Flow抽象类
public abstract class AbstractFlow<T> : Flow<T>, CancellableFlow<T> {
//实现Flow中的唯一接口
public final override suspend fun collect(collector: FlowCollector<T>) {
//注释1,简单封装
val safeCollector = SafeCollector(collector, coroutineContext)
try {
//注释2
collectSafely(safeCollector)
} finally {
safeCollector.releaseIntercepted()
}
}
public abstract suspend fun collectSafely(collector: FlowCollector<T>)
}
该抽象类实现了Flow接口,Flow接口只有唯一的collect方法,会在这里实现。
- 注释1处,collector就是调用collect方法时,传入的FlowCollector对象。
- 注释1处,coroutineContext是协程上下文,这是因为collect本身就是一个挂起函数,是挂起函数就肯定有Continuation对象,而这里不对外暴露Continuation对象,但是可以通过编译器优化,我们可以拿到上下文对象。
- 在封装为safeCollector后,就会调用collectSafely(safeCollector)方法,根据前面分析,就会执行上游的lambda中的操作。
即分析到现在,我们知道flow{}的lambda中,可以调用emit发射方法,而该对象就是这个safeCollector,我们称为上游的FlowCollector。
暂时先不分析SafeCollector,我们来看看下游终止操作符collect{},根据前面分析我们可知flow{}会创建一个SafeFlow的对象,所以我们可以调用其实现接口Flow的唯一方法collect:
public suspend fun collect(collector: FlowCollector<T>)
其实最开始的测试代码的写法,是经过简化的,其实效果如下:
//把collect简写复原
flow {
emit(1)
emit(2)
emit(3)
emit(4)
emit(5)
}
.collect(object: FlowCollector<Int>{
override suspend fun emit(value: Int) {
println(value)
}
})
这里使用collect方法创建了一个下游FlowCollector对象,而Flow中的数据是通过回调该下游FlowCollector对象的emit方法收集到。
再回到代码1的AbstractFlow中,是不是有一种恍然大悟的感觉:只有调用了collect方法,才会让上游的lambda执行,这也就是"冷"的表现。
那么还剩一个问题:在AbstractFlow的collect(collector: FlowCollector)方法中,通知下游数据是通过下游操作符的collector的emit方法,而发送数据是通过上游操作符的safeCollector的emit方法,这是如何结合起来的呢?
谜底就在SafeCollector类,该类定义:
//注释1,函数引用
private val emitFun =
FlowCollector<Any?>::emit as Function3<FlowCollector<Any?>, Any?, Continuation<Unit>, Any?>
//实现类,该类又实现了FlowCollector接口
internal actual class SafeCollector<T> actual constructor(
//终止操作符的FlowCollector
@JvmField internal actual val collector: FlowCollector<T>,
//协程上下文
@JvmField internal actual val collectContext: CoroutineContext
) : FlowCollector<T>, ContinuationImpl(NoOpContinuation, EmptyCoroutineContext), CoroutineStackFrame {
...
//注释2,上游操作符中的FlowCollector会调用的发射方法
override suspend fun emit(value: T) {
return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
try {
//发射数据
emit(uCont, value)
} catch (e: Throwable) {
lastEmissionContext = DownstreamExceptionElement(e)
throw e
}
}
}
//内部方法
private fun emit(uCont: Continuation<Unit>, value: T): Any? {
val currentContext = uCont.context
currentContext.ensureActive()
// This check is triggered once per flow on happy path.
val previousContext = lastEmissionContext
if (previousContext !== currentContext) {
checkContext(currentContext, previousContext, value)
}
completion = uCont
//注释3
return emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
}
...
}
这里省略了部分方法实现,只展示重要代码段,分析:
- 首先是注释2,由于SafeCollector是实现FlowCollector接口,所以注释2处的emit就是上游操作符中的emit方法,即flow{ emit(0) }就是调用该方法。
- 该方法中,会使用suspendCoroutineUninterceptedOrReturn来实现挂起函数,即发送数据,调用私有的内部emit方法。
- 在私有的emit方法的注释3处,会调用emit()方法,方法参数分别为:collector也就是下游操作符中的FlowCollector,value也就是发射的数据0,this是Continuation对象。
- 结合注释1处的定义,这里使用了函数引用,用到了挂起函数CPS的原理,即FlowCollector的emit挂起函数,其CPS后的函数引用就是Function3<FlowCollector<Any?>, Any?, Continuation<Unit>, Any?> ,表示意思是:参数分别是接收者FlowCollector,发射的值Any>,拼接在参数后面的Continuation对象,和返回值Any?。
所以这里注释3处,就是调用了下游操作符的FlowCollector的emit方法,这样也就可以完美收集到数据了。
这里我们可以总结一下:
1.下游调用了collect方法,传递下游FlowCollector对象,才会触发上游数据发射。
2.上游数据发射,即上游的FlowCollector调用emit发射的数据,会通过转换调用下游的FlowCollector的emit方法来接收数据。
这里第一点解释了Flow冷的原因,第二点解释了Flow懒惰的原因:一次只能发送和接收一个数据。
2.总结
这么一分析完,其实可以发现Flow还是非常简单的,实现思路就类似与Callbck传递,终止操作符collect{}设置Callback(FlowCollector),触发上游flow{}的Callback(FlowCollector)发射数据。
而中间操作符,也是一样的思想:触发上游,接收上游数据。本篇重点分析了filter{},会发现最后还是返回一个Flow对象,在fitler{}实现中会调用collect{}方法,然后调用emit方法。