前言

在前面文章中，咱们说过在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办法，这样也就能够完美收集到数据了。

这儿咱们能够总结一下：

下流调用了collect办法，传递下流FlowCollector目标，才会触发上游数据发射。
上游数据发射，即上游的FlowCollector调用emit发射的数据，会通过转换调用下流的FlowCollector的emit办法来接纳数据。

这儿第一点解说了Flow冷的原因，第二点解说了Flow懒散的原因：一次只能发送和接纳一个数据。

sequenceDiagram
下流->>上游: collect()
Note left of 上游:触发lambda履行
上游-->>上游:上游Collector.emit(0)
上游-->>上游:上游Collector.emitFun(0)
上游->>下流:下流Collector.emit(0)
Note right of 下流:收集到0

`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办法。

协程(23) | Flow原理解析

前言

正文

`Flow`原理探究

`Flow`中间操作符

总结

作者信息

协程(23) | Flow原理解析

前言

正文

Flow原理探究

Flow中间操作符

总结

相关文章

一次疑似 JVM native 内存泄漏的排查实录

Android Binder初始化

下一代智能合约编程语言Move（四）

货拉拉iOS模块化路由框架:TheRouterSwift

作者信息

`Flow`原理探究

`Flow`中间操作符