这一次，让Kotlin Flow 操作符真正好用起来

作者: 小鱼人爱编程 | 来源:发表于2023-04-27 11:46 被阅读0次

前言

Kotlin Flow 如此受欢迎大部分归功于其丰富、简洁的操作符，巧妙使用Flow操作符可以大大简化我们的程序结构，提升可读性与可维护性。
然而，虽然好用，但有些操作符不太好理解，可惜的是网上大部分文章只是简单介绍其使用，并没有梳理各个操作符的关系以及引入的缘由，本篇将通过关键原理与使用场景串联大部分操作符，以期达到举一反三的效果。
通过本篇文章，你将了解到：

操作符全家福

单Flow操作符的原理以及使用场景

单Flow操作符里的多协程原理以及使用场景

多Flow操作符里的多协程原理以及使用场景

Flow操作符该怎么学？

1. 操作符全家福

Flow操作符分类.png

红色部分为使用了多协程的操作符
上图仅包含常用官方提供的操作符，其它未包含进来的操作符原理也是类似的，当然我们也可以封装自己的操作符

由图上可知，将操作符分为了三类：

构建操作符

中间操作符

末端操作符

2. 单Flow操作符的原理以及使用场景

最简单的Flow

    fun test0() {
        runBlocking {
            //构造flow
            val flow = flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }
            //收集flow
            flow.collect {
                //下游
                println("collect:$it ${Thread.currentThread()}")
            }
        }
    }

如上包含了两种操作符：构造操作符flow与末端操作符collect。

image.png

总结来说，flow调用流程简化为：两个操作符+两个闭包+emit函数：

collect操作符触发调用，执行了flow的闭包

flow闭包里调用emit函数，执行了collect闭包

Flow返回集合

collect闭包里仅仅只是打印了数据，有个需求：需要将收集到的数据放在List里。
很容易就想到：

    fun test00() {
        runBlocking {
            val result = mutableListOf<String>()
            //构造flow
            val flow = flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }
            //收集flow
            flow.collect {
                //下游
                println("collect:$it ${Thread.currentThread()}")
                result.add(it)
            }
        }
    }

如上，定义List变量，在collect的闭包里收到数据后填充到List里。
某天，我们发现这个功能挺常用，需要将它封装起来，外界只需要传入List对象即可。

public suspend fun <T, C : MutableCollection<in T>> Flow<T>.toCollection(destination: C): C {
    collect { value ->
        destination.add(value)
    }
    return destination
}

外部使用：

    fun test01() {
        runBlocking {
            val result = mutableListOf<String>()
            flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }.toList(result)
        }
    }

如此一看，简单了许多，这也是官方提供的Flow操作符。

原理很简单：

作为Flow的扩展函数

重写了Flow的collect闭包，也就是FlowCollector的emit函数

后续很多操作符都是这么个套路，比如取Flow的第一个数据：first操作符，比如取对Flow里相邻的两个值做操作：reduce操作符等等。

Flow变换操作符

有个需求：在Flow流到下游之前，对数据进行处理，处理完成后再发射出去。
可以使用transform 操作符。

    fun test02() {
        runBlocking {
            flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }.transform {
                emit("$it man")
            }.collect {
                println("$it")
            }
        }
    }

再看看原理：

public inline fun <T, R> Flow<T>.transform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = flow { // Note: safe flow is used here, because collector is exposed to transform on each operation
    collect { value ->
        //上游的数据先经过transform处理
        return@collect transform(value)
    }
}

依然是Flow扩展函数，返回一个新的Flow对象

新Flow对象重写了flow闭包，该闭包里调用collect收集了原始Flow的数据

当数据到来后，经过transform处理，而我们自定义的transform闭包里将数据再次发射出去

最后新返回的flow的collect闭包被调用

上面只是使用了一个transform操作符，若是多个transform操作符，该怎么去分析呢？其实，套路是有迹可循的。
这里涉及到了一种设计模式：装饰者模式

image.png

每调用1个transform操作符就会新生成一个Flow对象，该对象装饰了它的上一个(扩展)对象，如上Flow1装饰原始Flow，Flow2装饰Flow1。

    fun test02() {
        runBlocking {
            flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }.transform {
                emit("$it 1")
            }.transform {
                emit("$it 2")
            }.transform {
                emit("$it 3")
            }.collect {
                println("$it")
            }
        }
    }

如上，相信你很快就知道输出结果了。

你可能觉得transform还需要自己发射数据，有点麻烦，map可解君忧。

    fun test03() {
        runBlocking {
            flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
            }.map {
                "$it 1"
            }.collect {
                println("$it")
            }
        }
    }

map内部封装了transform。

过滤操作符

有个需求：对上流的数据进行某种条件的筛选过滤。
有了transform的经验，我们很容易想到定义扩展函数返回新的Flow，并重写collect的闭包，在闭包里进行限制。

public inline fun <T> Flow<T>.filter(crossinline predicate: suspend (T) -> Boolean): Flow<T> = transform { value ->
    //条件满足再发射
    if (predicate(value)) return@transform emit(value)
}

internal inline fun <T, R> Flow<T>.unsafeTransform(
    @BuilderInference crossinline transform: suspend FlowCollector<R>.(value: T) -> Unit
): Flow<R> = unsafeFlow { // Note: unsafe flow is used here, because unsafeTransform is only for internal use
    collect { value ->
        return@collect transform(value)
    }
}

使用方式：

    fun test04() {
        runBlocking {
            flow {
                //上游
                emit("hello world ${Thread.currentThread()}")
                emit("fish")
            }.filter {
                //包含hello字符串才继续往下发送
                it.contains("hello")
            }.collect {
                println("$it")
            }
        }
    }

掌握了以上套路，再去理解其它类似的操作符就很简单了，都是一些简单的变种。

3. 单Flow操作符里的多协程原理以及使用场景

Flow里如何切换协程与线程

上面提到的操作符，如map、filter，相信大家也看出来了：

整个流程的过程没有涉及到其它协程，也没有涉及到其它的线程，是比较单纯也比较容易理解

有个需求：在主线程执行collect操作符，在flow闭包里执行耗时操作。
此时我们就需要flow闭包里的代码在子线程执行。
你可能一下子就说出了答案：使用flowOn操作符。

    fun test05() {
        runBlocking {
            flow {
                //上游
                println("emit ${Thread.currentThread()}")
                emit("hello world")
            }.flowOn(Dispatchers.IO)//flowOn 之前的操作符在新协程里执行
                .collect {
                    println("$it")
                    println("collect ${Thread.currentThread()}")
                }
        }
    }
//打印结果
emit Thread[DefaultDispatcher-worker-1 @coroutine#3,5,main]
hello world
collect Thread[main @coroutine#2,5,main]

可以看出，flow闭包(上游)，collect闭包(下游)分别执行在不同的协程以及不同的线程里。
flowOn原理简单来说：

构造了新的协程执行flow闭包，又因为指定了协程分发器为Dispatchers.IO，因此会在子线程里执行flow闭包
原理是基于ChannelFlow

Flow处理背压

有个需求：上游发射数据速度高于下游，如何提升发射效率？
如下：

    fun test06() {
        runBlocking {
            val time = measureTimeMillis {
                flow {
                    //上游
                    println("emit ${Thread.currentThread()}")
                    emit("hello world")
                    delay(1000)
                    emit("hello world2")
                }.collect {
                        delay(2000)
                        println("$it")
                        println("collect ${Thread.currentThread()}")
                    }
            }
            println("use time:$time")
        }
    }
//打印
emit Thread[main @coroutine#2,5,main]
hello world
collect Thread[main @coroutine#2,5,main]
hello world2
collect Thread[main @coroutine#2,5,main]
use time:5024

使用buffer操作符解决背压问题：

    fun test06() {
        runBlocking {
            val time = measureTimeMillis {
                flow {
                    //上游
                    println("emit ${Thread.currentThread()}")
                    emit("hello world")
                    delay(1000)
                    emit("hello world2")
                }.buffer().collect {
                        delay(2000)
                        println("$it")
                        println("collect ${Thread.currentThread()}")
                    }
            }
            println("use time:$time")
        }
    }
//打印结果
emit Thread[main @coroutine#3,5,main]
hello world
collect Thread[main @coroutine#2,5,main]
hello world2
collect Thread[main @coroutine#2,5,main]
use time:4065

可以看出，总耗时减少了。
buffer原理简单来说：

构造了新的协程执行flow闭包，上游数据会发送到Channel 缓冲区里，发送完成继续发送下一条
collect操作符监听缓冲区是否有数据，若有则收集成功
原理是基于ChannelFlow

关于flowOn和buffer更详细的原理请移步：Kotlin Flow 背压和线程切换竟然如此相似

上游覆盖旧数据

有个需求：上游生产速度很快，下游消费速度慢，我们只关心最新数据，旧的数据没价值可以丢掉。
使用conflate操作符处理：

    fun test07() {
        runBlocking {
            flow {
                //上游
                repeat(5) {
                    emit("emit $it")
                    delay(100)
                }
            }.conflate().collect {
                delay(500)
                println("$it")
            }
        }
    }
//打印结果：
emit 0
emit 4

可以看出，中间产生的数据由于下游没有来得及消费，被上游新的数据冲刷掉了。

conflate原理简单来说：

相当于使用了buffer操作符，该buffer只能容纳一个数据，新来的数据将会覆盖旧的数据
原理是基于ChannelFlow

Flow变换取最新值

有个需求：在使用transform处理数据的时候，若是它处理比较慢，当有新的值过来后就取消未处理好的值。
使用transformLatest操作符处理：

    fun test08() {
        runBlocking {
            flow {
                //上游，协程1
                repeat(5) {
                    emit("emit $it")
                }
                println("emit ${Thread.currentThread()}")
            }.transformLatest {
                //协程2
                delay(200)
                emit("$it fish")
            }.collect {
                println("collect ${Thread.currentThread()}")
                println("$it")
            }
        }
    }
打印结果：
emit Thread[main @coroutine#3,5,main]
collect Thread[main @coroutine#2,5,main]
emit 4 fish

可以看出，由于transform处理速度比较慢，上游有新的数据过来后会取消transform里未处理的数据。
查看源码是如何处理的：

override suspend fun flowCollect(collector: FlowCollector<R>) {
    coroutineScope {
        var previousFlow: Job? = null
        //开始收集上游数据
        flow.collect { value ->
            previousFlow?.apply {
                //若是之前的协程还在，则取消
                cancel(ChildCancelledException())
                join()
            }
            //开启协程执行，此处选择不分发新线程
            previousFlow = launch(start = CoroutineStart.UNDISPATCHED) {
                collector.transform(value)
            }
        }
    }
}

transformLatest原理简单来说：

构造新的协程1执行flow闭包，收集到数据后再开启新的协程2，在协程里会调用transformLatest的闭包，最终调用collect的闭包
协程1继续发送数据，若是发现协程2还在运行，则取消协程2
原理是基于ChannelFlow

同理，map也有类似的操作符：

    fun test09() {
        runBlocking {
            flow {
                //上游
                repeat(5) {
                    emit("emit $it")
                }
                println("emit ${Thread.currentThread()}")
            }.mapLatest {
                delay(200)
                "$it fish"
            }.collect {
                println("collect ${Thread.currentThread()}")
                println("$it")
            }
        }
    }
//打印结果
emit Thread[main @coroutine#3,5,main]
collect Thread[main @coroutine#2,5,main]
emit 4 fish

收集最新的数据

有个需求：监听下载进度，UI展示最新进度。
分析：此种场景下，我们只是关注最新的进度，没必要频繁刷新UI，因此使用Flow实现时上游发射太快了可以忽略旧的数据。
使用collectLatest操作符实现：

    fun test014() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    repeat(100) {
                        emit(it + 1)
                    }
                }
                flow1.collectLatest {
                    delay(20)
                    println("collect progress $it")
                }
            }
            println("use time:$time")
        }
    }
//打印结果
collect progress 100
use time:169

collectLatest原理简单来说：

开启新协程执行flow闭包
若是collect收集比较慢，下一个数据emit过来后会取消未处理的数据
原理是基于ChannelFlow

4. 多Flow操作符里的多协程原理以及使用场景

很多时候我们不止操作单个Flow，有可能需要结合多个Flow来实现特定的业务场景。

展平流

flatMapConcat

有个需求：请求某个学生的班主任信息，这里涉及到两个接口：

请求学生信息，使用Flow1表示

请求该学生的班主任信息，使用Flow2表示

我们需要先拿到学生的信息，通过信息里带的班主任id去请求班主任信息

分析需求可知：获取学生信息的请求和获取班主任信息的请求是串行的，有前后依赖关系。
使用flatMapConcat操作符实现：

    fun test010() {
        runBlocking {
            val flow1 = flow {
                emit("stuInfo")
            }
            flow1.flatMapConcat {
                //flow2
                flow {
                    emit("$it teachInfo")
                }
            }.collect {
                println("collect $it")
            }
        }
    }
//打印结果：
collect stuInfo teachInfo

从打印结果可以看出：

所谓展平，实际上就是将两个Flow的数据拍平了输出

当然，你也可以请求多个学生的班主任信息：

    fun test011() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    println("emit ${Thread.currentThread()}")
                    emit("stuInfo 1")
                    emit("stuInfo 2")
                    emit("stuInfo 3")
                }
                flow1.flatMapConcat {
                    //flow2
                    flow {
                        println("flatMapConcat ${Thread.currentThread()}")
                        emit("$it teachInfo")
                        delay(1000)
                    }
                }.collect {
                    println("collect ${Thread.currentThread()}")
                    println("collect $it")
                }
            }
            println("use time:$time")
        }
    }
//打印结果：
emit Thread[main @coroutine#2,5,main]
flatMapConcat Thread[main @coroutine#2,5,main]
collect Thread[main @coroutine#2,5,main]
collect stuInfo 1 teachInfo
flatMapConcat Thread[main @coroutine#2,5,main]
collect Thread[main @coroutine#2,5,main]
collect stuInfo 2 teachInfo
flatMapConcat Thread[main @coroutine#2,5,main]
collect Thread[main @coroutine#2,5,main]
collect stuInfo 3 teachInfo
use time:3032

flatMapConcat原理简单来说：

flatMapConcat 并没有涉及到多协程，使用了装饰者模式
先将Flow2使用map进行变换，而后将Flow1、Flow2数据发射出来
Concat顾名思义，将两个Flow连接起来

flatMapMerge

有个需求：在flatMapConcat里，先查询了学生1的班主任信息后才会查询学生2的班主任信息，依照此顺序进行查询。现在需要提升效率，同时查询多个多个学生的班主任信息。
使用flatMapMerge操作符实现：

    fun test012() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    println("emit ${Thread.currentThread()}")
                    emit("stuInfo 1")
                    emit("stuInfo 2")
                    emit("stuInfo 3")
                }
                flow1.flatMapMerge(4) {
                    //flow2
                    flow {
                        println("flatMapMerge ${Thread.currentThread()}")
                        emit("$it teachInfo")
                        delay(1000)
                    }
                }.collect {
                    println("collect ${Thread.currentThread()}")
                    println("collect $it")
                }
            }
            println("use time:$time")
        }
    }
//打印结果：
flatMapMerge Thread[main @coroutine#6,5,main]
collect Thread[main @coroutine#2,5,main]
collect stuInfo 1 teachInfo
collect Thread[main @coroutine#2,5,main]
collect stuInfo 2 teachInfo
collect Thread[main @coroutine#2,5,main]
collect stuInfo 3 teachInfo
use time:1086

可以看出，flatMapMerge由于是并发执行，整体速度比flatMapConcat快了很多。
flatMapMerge可以指定并发的数量，当指定flatMapMerge(0)时，flatMapMerge退化为flatMapConcat。
关键源码如下：

override suspend fun collectTo(scope: ProducerScope<T>) {
    val semaphore = Semaphore(concurrency)
    val collector = SendingCollector(scope)
    val job: Job? = coroutineContext[Job]
    flow.collect { inner ->
        job?.ensureActive()
        //并发数限制锁
        semaphore.acquire()
        scope.launch {
            //开启新的协程
            try {
                //执行flatMapMerge闭包里的flow
                inner.collect(collector)
            } finally {
                semaphore.release() // Release concurrency permit
            }
        }
    }
}

flatMapMerge原理简单来说：

flow1里的每个学生信息会触发去获取班主任信息flow2
新开了协程去执行flow2的闭包
原理是基于ChannelFlow

flatMapLatest

有个需求：flatMapConcat 是线性执行的，可以使用flatMapMerge提升效率。为了节约资源，在请求班主任信息的时候，若是某个学生的班主任信息没有返回，而下一个学生的班主任信息已经开始请求，则取消上一个没有返回的班主任Flow。
使用flatMapLatest操作符实现：

    fun test013() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
//                    println("emit ${Thread.currentThread()}")
                    emit("stuInfo 1")
                    emit("stuInfo 2")
                    emit("stuInfo 3")
                }
                flow1.flatMapLatest {
                    //flow2
                    flow {
//                        println("flatMapLatest ${Thread.currentThread()}")
                        delay(1000)
                        emit("$it teachInfo")
                    }
                }.collect {
//                    println("collect ${Thread.currentThread()}")
                    println("collect $it")
                }
            }
            println("use time:$time")
        }
    }
//打印结果：
collect stuInfo 3 teachInfo
use time:1105

可以看出，只有学生3的班主任信息打印出来了，并且整体时间都减少了。
flatMapLatest原理简单来说：

和transformLatest很相似
原理是基于ChannelFlow

简单总结一下关于收集最新数据的操作符：

transformLatest、mapLatest、collectLatest、flatMapLatest 四者的核心实现都是ChannelFlowTransformLatest，而它最终继承自：ChannelFlow

组合流

combine

有个需求：查询学生的性别以及选修了某个课程。
分析：涉及到两个需求，查询学生性别与查询选修课程，输出结果是：性别：xx，选修了：xx课程。这俩请求可以同时发出，并没有先后顺序，因此我们没必要使用flatMapXX系列操作符。
使用combine操作符：

    fun test015() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    emit("stuSex 1")
                    emit("stuSex 2")
                    emit("stuSex 3")
                }
                val flow2 = flow {
                    emit("stuSubject")
                }
                flow1.combine(flow2) {
                    sex, subject->"$sex-->$subject"
                }.collect {
                    println(it)
                }
            }
            println("use time:$time")
        }
    }
//打印结果：
stuSex 1-->stuSubject
stuSex 2-->stuSubject
stuSex 3-->stuSubject
use time:46

可以看出，flow1的每个emit和flow2的emit关联起来了。
combine操作符有个特点：

短的一方会等待长的一方结束后才结束

看个例子就比较清晰：

    fun test016() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    emit("a")
                    emit("b")
                    emit("c")
                    emit("d")
                }
                val flow2 = flow {
                    emit("1")
                    emit("2")
                }
                flow1.combine(flow2) {
                        sex, subject->"$sex-->$subject"
                }.collect {
                    println(it)
                }
            }
            println("use time:$time")
        }
    }
//打印结果
a-->1
b-->2
c-->2
d-->2
use time:45

flow2早就发射到"2"了，会一直等到flow1发射结束。

combine原理简单来说：

image.png

zip

在combine需求的基础上，我们又有个优化：无论是学生性别还是学生课程，只要某个Flow获取结束了就取消Flow。
使用zip操作符：

    fun test017() {
        runBlocking {
            val time = measureTimeMillis {
                val flow1 = flow {
                    emit("a")
                    emit("b")
                    emit("c")
                    emit("d")
                }
                val flow2 = flow {
                    emit("1")
                    emit("2")
                }
                flow1.zip(flow2) {
                        sex, subject->"$sex-->$subject"
                }.collect {
                    println(it)
                }
            }
            println("use time:$time")
        }
    }
//打印结果
a-->1
b-->2
use time:71

可以看出flow2先结束了，并且flow1没发送完成。
zip原理简单来说：

image.png

可以看出，zip的特点：

短的Flow结束，另一个Flow也结束

5. Flow操作符该怎么学？

以上我们由浅入深分别分析了：

单个Flow操作符原理与使用场景

单个Flow操作符切换多个协程的原理与使用场景

多个Flow操作符切换多个协程的原理与使用场景

以上三者是递进关系，第1点比较简单，第2点难度适中。
尤其是第3点比较难以理解，因为涉及到了其它的知识：Channel、ChannelFlow、多协程、线程切换等。
在之前的文章中有提到过：ChannelFlow是Flow复杂操作符的基础，想要掌握复杂操作符的原理需要明白ChannelFlow的运行机制，有兴趣可移步：当，Kotlin Flow与Channel相逢

建议Flow操作符学习步骤：

先会使用简单的操作符filter、map等

再学会使用flowOn、buffer、callbackFlow等操作符

进而使用flatMapXXX以及combine、zip等操作符

最后可以看看其实现原理，达到举一反三应用到实际需求里

Flow操作符的闭坑指南：

涉及到多协程的操作符，需要关注其执行的线程环境

涉及到多协程的操作符，需要关注协程的生命周期

说实话，Flow操作符要掌握好挺难的，它几乎涉及了协程所有的知识点，也是协程实际应用的精华。这篇是我在协程系列里花费时间最长的文章了（也许也是最后一篇了），即使自己弄明白了，怎样把它很自然地递进引出也是个有挑战的事。
若你能够在本篇的分析中得到一点启发，那说明我的分享是有价值的。
由于篇幅关系，一些操作符debounce、sample等并没有分析，也没有再贴flatMapXXX的源码细节（这部分之前的文章都有分析过），若你有需要可以给我留言评论。

本文基于Kotlin 1.6.1，覆盖所有Flow操作符的demo

您若喜欢，请点赞、关注、收藏，您的鼓励是我前进的动力

持续更新中，和我一起步步为营系统、深入学习Android/Kotlin

1、Android各种Context的前世今生
2、Android DecorView 必知必会
3、Window/WindowManager 不可不知之事
4、View Measure/Layout/Draw 真明白了
5、Android事件分发全套服务
6、Android invalidate/postInvalidate/requestLayout 彻底厘清
7、Android Window 如何确定大小/onMeasure()多次执行原因
8、Android事件驱动Handler-Message-Looper解析
9、Android 键盘一招搞定
10、Android 各种坐标彻底明了
11、Android Activity/Window/View 的background
12、Android Activity创建到View的显示过
13、Android IPC 系列
14、Android 存储系列
15、Java 并发系列不再疑惑
16、Java 线程池系列
17、Android Jetpack 前置基础系列
18、Android Jetpack 易懂易学系列
19、Kotlin 轻松入门系列
20、Kotlin 协程系列全面解读

这一次，让Kotlin Flow 操作符真正好用起来

前言

1. 操作符全家福

2. 单Flow操作符的原理以及使用场景

最简单的Flow

Flow返回集合

Flow变换操作符

过滤操作符

3. 单Flow操作符里的多协程原理以及使用场景

Flow里如何切换协程与线程

Flow处理背压

上游覆盖旧数据

Flow变换取最新值

收集最新的数据

4. 多Flow操作符里的多协程原理以及使用场景

展平流

flatMapConcat

flatMapMerge

flatMapLatest

组合流

combine

zip

5. Flow操作符该怎么学？

您若喜欢，请点赞、关注、收藏，您的鼓励是我前进的动力

持续更新中，和我一起步步为营系统、深入学习Android/Kotlin

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Android开发