我正在参加「·启航计划」
前语
协程系列文章：
- 一个小故事讲了解进程、线程、Kotlin 协程到底啥联系？
- 少年，你可知 Kotlin 协程最初的姿态？
- 讲真，Kotlin 协程的挂起/恢复没那么奥秘(故事篇)
- 讲真，Kotlin 协程的挂起/恢复没那么奥秘(原理篇)
- Kotlin 协程调度切换线程是时分解开真相了
- Kotlin 协程之线程池探索之旅(与Java线程池PK)
- Kotlin 协程之撤销与反常处理探索之旅(上)
- Kotlin 协程之撤销与反常处理探索之旅(下)
- 来，跟我一同撸Kotlin runBlocking/launch/join/async/delay 原理&运用
- 持续来，同我一同撸Kotlin Channel 深水区
- Kotlin 协程 Select：看我怎样多路复用
- Kotlin Sequence 是时分派上用场了
- Kotlin Flow啊，你将流向何方？
- Kotlin Flow 背压和线程切换竟然如此类似
- Kotlin SharedFlow&StateFlow 热流到底有多热？
- 狂飙吧，Lifecycle与协程、Flow的化学反应
- 来吧！承受Kotlin 协程–线程池的7个魂灵拷问
- 当，Kotlin Flow与Channel相逢
- 这一次，让Kotlin Flow 操作符真正好用起来
Kotlin Flow 如此受欢迎大部分归功于其丰厚、简洁的操作符，巧妙运用Flow操作符能够大大简化咱们的程序结构，提高可读性与可维护性。
然而，尽管好用，但有些操作符不太好了解，惋惜的是网上大部分文章仅仅简略介绍其运用，并没有梳理各个操作符的联系以及引进的缘由，本篇将经过要害原理与运用场景串联大部分操作符，以期到达触类旁通的作用。
经过本篇文章，你将了解到：
- 操作符全家福
- 单Flow操作符的原理以及运用场景
- 单Flow操作符里的多协程原理以及运用场景
- 多Flow操作符里的多协程原理以及运用场景
- Flow操作符该怎样学？
1. 操作符全家福
红色部分为运用了多协程的操作符
上图仅包括常用官方供给的操作符，其它未包括进来的操作符原理也是类似的，当然咱们也能够封装自己的操作符
由图上可知，将操作符分为了三类：
- 构建操作符
- 中心操作符
- 末端操作符
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。
总结来说，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操作符，该怎样去剖析呢？其实，套路是有迹可循的。
这儿触及到了一种设计形式：装修者形式
每调用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原理简略来说：
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原理简略来说：
能够看出，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 协程系列全面解读
