前言

在前面文章中，我们说过在Flow出来之前，Kotlin的协程可能还不足以有绝对优势，在Flow发布之后，协程才真正被大家认可。

在文章 # 协程(10) | Flow 中，我们介绍了Flow的简单使用，那么本篇文章就来看看Flow的原理。

正文

区别于上一篇文章所说的Channel是热的，Flow的最大特点就是"冷"，这里的"冷"还包括"懒"，即当Flow有终止操作符时，上游才开始发射数据，且一次只发射一个。

Flow原理探究

我们还是以最简单的测试代码入手：

private suspend fun testFlow() {
    //上游操作符，创建Flow的同时，发射数据
    flow {
        emit(1)
        emit(2)
        emit(3)
        emit(4)
        emit(5)
    }
        //终止操作符，接收数据
        .collect {
            logX(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修饰，内部使用
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) {
            logX(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懒惰的原因：一次只能发送和接收一个数据。

Flow中间操作符

Flow的强大之处不仅仅是"冷"的特性，还因为其有方便的中间操作符。

根据前面的思想：下游操作符触发上游操作符的动作执行，上游操作符再把数据传递给下游操作符，我们是否可以设想一下中间操作符的运作规则。

比如下面代码：

private suspend fun testFlow() {
    flow {
        emit(1)
        emit(2)
        emit(3)
        emit(4)
        emit(5)
    }
        .filter { it > 2 }
        .map { it * 2 }
        .collect { value -> logX(value) }
}

由于还是"冷"的特性，collect{}会触发map{}的执行，map{}会触发filter{}的执行，filter{}会触发flow{}的执行，然后flow{}把数据传递给filter{}，filter{}把数据传递给map{}，最后map{}把数据传递给collect{}收集到。

这个流程还是符合Flow的"冷"的特性，也符合"懒"的特性，因为每次只处理一个数据。我们就以filter为例，来看看其实现代码：

//inline函数，参数为crossinline类型，返回值类型为Flow，调用transform函数
public inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T> 
 = transform { value ->
    if (predicate(value)) return@transform emit(value)
}

//这里会调用unsafeTransform，而不是transform函数
//这里会继续调用unsafeFlow函数
internal inline fun <T, R> Flow<T>.unsafeTransform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = unsafeFlow { //注释2
    collect { value ->
        return@collect transform(value)
    }
}

//会返回一个Flow对象
internal inline fun <T> unsafeFlow(@BuilderInference crossinline 
    block: suspend FlowCollector<T>.() -> Unit): Flow<T> {
    return object : Flow<T> {
        //注释1
        override suspend fun collect(collector: FlowCollector<T>) {
            collector.block()
        }
    }
}

上述代码看起来平平无奇，但是如果对Kotlin的高阶函数理解不透彻的话，还是很难理解，我们就来一一分析：

• 首先看最后的unsafeFlow函数，通过object返回一个Flow对象，这时如果在filter{}后调用collect{}函数时，这里就会触发注释1出的方法执行。
• 在该方法中，重点是collector.block()，这里会触发代码块block()的执行，这里的block代码块的类型是suspend FlowCollector.() -> Unit类型，是FlowCollector的成员对象挂起函数。
• 该block就是unsafeTransform中的lambda代码块，即注释2中4行lambda代码块。
• 在注释2的lambda中，会调用collect方法，这是因为该方法是Flow的扩展函数，这也就会导致上游操作符执行，即flow{}.filter{}.collect{}中flow{}将会执行，而这其中value就是发射出来的值。
• 注释2中的value，会经过其参数transform(value)处理，这个参数就是if(predicate(value)) emit(value)，即flow{}发射出的数据，再经过filter{}后，重新调用emit()方法。
• 看到这个emit()方法，其实也就是回到了上面逻辑了，这个就是下游collect{}的上游FlowCollector了。

这里可以看出，调用中间操作符filter{}会创建出新的Flow对象，而且会对数据重新进行发射。

总结

这么一分析完，其实可以发现Flow还是非常简单的，实现思路就类似与Callbck传递，终止操作符collect{}设置Callback(FlowCollector)，触发上游flow{}的Callback(FlowCollector)发射数据。

而中间操作符，也是一样的思想：触发上游，接收上游数据。本篇重点分析了filter{}，会发现最后还是返回一个Flow对象，在fitler{}实现中会调用collect{}方法，然后调用emit方法。

作者：元浩875
链接：https://juejin.cn/post/7173494906319536142