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方法。