关于Flow的原理解析

贡念

2023-12-01

Flow是基于协程实现的异步数据流，所以在学习flow原理之前需要掌握协程相关的知识。

话不多说，直接开摆，不是，直接开始。

首先最简单的创建flow我们都知道flow{ emit(vale) },这里使用lambda直接将block传入flow里面了，我们都知道flow是冷流，只有当我们collect收集flow的时候，这里面的block才会执行，这么说没人反对吧，但是这是怎么实现的，其实就是巧妙的应用了协程的挂起函数。

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

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

可以看到我们flow{}返回的是一个SafeFlow()对象，代码块block继承的是挂起的一个接口FlowCollector,这里是不是想起来launch源码里面的一个参数block: suspend CoroutineScope.() -> Unit，没错就是一样的！

public interface FlowCollector<in T> {

    /**
     * Collects the value emitted by the upstream.
     * This method is not thread-safe and should not be invoked concurrently.
     */
    public suspend fun emit(value: T)
}

这个FlowCollector很重要，官方给的释义是：FlowCollector用作流的中间或终端收集器，表示接受流发出的值的实体。此接口通常不应直接实现，而应在实现自定义运算符时用作流生成器中的接收器。此接口的实现不是线程安全的。这个接口只有一个emit方法，这个方法我们再熟悉不过了，就是我们往流里面塞数据用的方法，可以看到emit是一个挂起函数。（这里可以类比一下liveData的协程实现方式）

再看下SafeFlow里面重写了一个collectSafely方法，具体实现看下SafeFlow继承的AbstractFlow类

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

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

    /**
     * Accepts the given [collector] and [emits][FlowCollector.emit] values into it.
     *
     * A valid implementation of this method has the following constraints:
     * 1) It should not change the coroutine context (e.g. with `withContext(Dispatchers.IO)`) when emitting values.
     *    The emission should happen in the context of the [collect] call.
     *    Please refer to the top-level [Flow] documentation for more details.
     * 2) It should serialize calls to [emit][FlowCollector.emit] as [FlowCollector] implementations are not
     *    thread-safe by default.
     *    To automatically serialize emissions [channelFlow] builder can be used instead of [flow]
     *
     * @throws IllegalStateException if any of the invariants are violated.
     */
    public abstract suspend fun collectSafely(collector: FlowCollector<T>)
}

AbstractFlow里面就能看到collect的实现方法，可以看到AbstractFlow继承的就是Flow基类，Flow基类中没什么实现方法，只有一个collect的抽象方法，所以我们着重看下AbstractFlow里面的collect是怎么实现异步执行emit的。

public suspend fun Flow<*>.collect(): Unit = collect(NopCollector)

public suspend inline fun <T> Flow<T>.collect(crossinline action: suspend (value: T) -> Unit): Unit =
    collect(object : FlowCollector<T> {
        override suspend fun emit(value: T) = action(value)
    })

无论使用collect()还是collect{}调用的都是Flow的collect抽象方法，collect真正实现还是在AbstractFlow里面，这里只是将collect的block封装成了FlowCollector（重写emit方法用于执行block)

封装的FlowCollector和CoroutineContext继续被封装为SafeCollector类型，并将该实例传入collectSafely方法用于在SafeFlow中实现block的执行。

@Suppress("UNCHECKED_CAST")
private val emitFun =
    FlowCollector<Any?>::emit as Function3<FlowCollector<Any?>, Any?, Continuation<Unit>, Any?>

@Suppress("CANNOT_OVERRIDE_INVISIBLE_MEMBER", "INVISIBLE_MEMBER", "INVISIBLE_REFERENCE", "UNCHECKED_CAST")
internal actual class SafeCollector<T> actual constructor(
    @JvmField internal actual val collector: FlowCollector<T>,
    @JvmField internal actual val collectContext: CoroutineContext
) : FlowCollector<T>, ContinuationImpl(NoOpContinuation, EmptyCoroutineContext), CoroutineStackFrame {

......

    /**
     *这是状态机重用的巧妙实现。首先，它检查它是否未同时使用（我们明确禁止），然后只缓存一个完成实例，
     *避免在每个发射上进行额外分配，使其在其热路径上有效地无垃圾。
     */
    override suspend fun emit(value: T) {
        return suspendCoroutineUninterceptedOrReturn sc@{ uCont ->
            try {
                emit(uCont, value)
            } catch (e: Throwable) {
                // 保存已引发emit异常（甚至检查上下文）的事实
                lastEmissionContext = DownstreamExceptionElement(e)
                throw e
            }
        }
    }

    private fun emit(uCont: Continuation<Unit>, value: T): Any? {
        val currentContext = uCont.context
        currentContext.ensureActive()
        // 检查flow是否在同一个上下文中
        val previousContext = lastEmissionContext
        if (previousContext !== currentContext) {
            checkContext(currentContext, previousContext, value)
        }
        completion = uCont
        return emitFun(collector as FlowCollector<Any?>, value, this as Continuation<Unit>)
    }

//检查context
private fun checkContext(
    currentContext: CoroutineContext,
    previousContext: CoroutineContext?,
    value: T
) {
    if (previousContext is DownstreamExceptionElement) {
        exceptionTransparencyViolated(previousContext, value)
    }
    checkContext(currentContext)
    lastEmissionContext = currentContext
}

通过上面的分析可知，未调用collect之前是不会调用collectSafely方法不会收集我们塞入emit中的数据，只有调用collect方法才会调用到，进而以扩展函数的形式调用到flow的block，block里面就是我们在emit里面写入的数据，所以作为接收器的SafeCollector调用了emit方法。

FlowCollector的block在SafeCollector中以value的形式进行传递，最后返回一个emitFun变量，该变量的类型就是FlowCollector的emit的block类型。

看到这个emitFun的参数会发现怎么多了一个，因为emit是挂起函数默认有一个参数的传递就是Continuation，这一块kotlin替我们做了，但是对于java来说必须要传递Continuation进去。通过这种方式传入一样的Continuation，保证了Continuation的统一。

说了这么多总结一下：

利用扩展函数的性质，调用到flow的block进而调用了SafeCollector的emit，而这里的emit会调用到传进来的FlowCollector的emit，而传进来的emit函数被重写调用block，所以就会调用到collect的block。因为只有调用collect之后进而调用到safeFlow的collect函数，进而才会调用到collectSafely函数去执行flow的代码。所以不调用collect的话，flow的代码构建块是不会执行的，最多返回一个safeFlow的对象而已。

作者：Lu_Hsiang
链接：https://juejin.cn/post/7120112404393885733

关于Flow的原理解析

相关阅读

相关文章

相关问答

相关文档