【Koltin Flow（三）】Flow操作符之中间操作符（二）

目录

【Koltin Flow（一）】五种创建flow的方式
【Koltin Flow（二）】Flow操作符之末端操作符
【Koltin Flow（三）】Flow操作符之中间操作符（一）
【Koltin Flow（三）】Flow操作符之中间操作符（二）
【Koltin Flow（三）】Flow操作符之中间操作符（三）
【Koltin Flow（四）】Flow背压
【Koltin Flow（五）】SharedFlow及StateFlow

前言

1. 本篇主要介绍的有组合操作符、回调操作符，也会涉及一些其他的操作符。

组合操作符

1. zip zip主要实现组合的功能，将两个flow一一组合发出，其中一个结束，则zip结束。

代码如下：

       val flow1  = listOf(1,2).asFlow()
            val flow2 = listOf("one","two","three").asFlow()
            flow1.zip(flow2){value1,value2->
                "zip $value1,$value2"
            }.collect {
                Log.d(TAG.TAG,it)
            }

日志如下：

2022-08-01 11:29:49.937 4576-4606/edu.test.demo D/Test-TAG: zip 1,one
2022-08-01 11:29:49.937 4576-4606/edu.test.demo D/Test-TAG: zip 2,two

分析：

• 可以看出zip将两个flow的值进行了一一组合。
• 另外也可以看出只打印出了两对值，因为第一个flow只有两个值，所以在两对结束之后zip已经结束。

2. merge 是将两个flow合并起来，将每个值依次发出来

代码如下：

            val flow1  = listOf(1,2).asFlow()
            val flow2 = listOf("one","two","three").asFlow()
            merge(flow1,flow2).collect {value->
                Log.d(TAG.TAG,value.toString())
            }

日志如下：

2022-08-01 11:33:14.765 4636-4662/edu.test.demo D/Test-TAG: 1
2022-08-01 11:33:14.765 4636-4662/edu.test.demo D/Test-TAG: 2
2022-08-01 11:33:14.765 4636-4662/edu.test.demo D/Test-TAG: one
2022-08-01 11:33:14.765 4636-4662/edu.test.demo D/Test-TAG: two
2022-08-01 11:33:14.765 4636-4662/edu.test.demo D/Test-TAG: three

分析：

• 可以看出merge在将flow1和flow2合并之后将五个值依次发送出来。

3. combine、combineTransform，combine是组合每个流最新的值，combineTransform为combine+transform。

代码如下：

            val flow1  = flow {
                repeat(3){
                    delay(50)
                    emit(it)
                }
            }
            val flow2 = flow<String> {
                delay(100)
                emit("one")
                delay(100)
                emit("two")
            }
            flow1.combine(flow2){value1,value2->
                "combine $value1,$value2"
            }.collect {
                Log.d(TAG.TAG,it)
            }
            flow1.combineTransform(flow2){value1,value2->
                emit("combineTransform $value1,$value2")
            }.collect {
                Log.d(TAG.TAG,it)
            }

日志如下：

2022-08-01 11:47:13.049 5525-5552/edu.test.demo D/Test-TAG: combine 0,one
2022-08-01 11:47:13.063 5525-5553/edu.test.demo D/Test-TAG: combine 1,one
2022-08-01 11:47:13.114 5525-5552/edu.test.demo D/Test-TAG: combine 2,one
2022-08-01 11:47:13.152 5525-5553/edu.test.demo D/Test-TAG: combine 2,two
2022-08-01 11:47:13.256 5525-5551/edu.test.demo D/Test-TAG: combineTransform 0,one
2022-08-01 11:47:13.258 5525-5551/edu.test.demo D/Test-TAG: combineTransform 1,one
2022-08-01 11:47:13.310 5525-5551/edu.test.demo D/Test-TAG: combineTransform 2,one
2022-08-01 11:47:13.357 5525-5551/edu.test.demo D/Test-TAG: combineTransform 2,two

分析：

• 可以看出combine将flow1与flow2的值进行了组合，但是可以看到two只组合打印了一次，这是因为在0,1,2 发送的时候，two都还没发送，所以flow2的最新值都是one。
• combineTransform 可以实现类似的效果，也可以进行其他的操作，具体可以参照transform。

4. flattenConcat、flattenMerge，flattenConcat将多个flow展平，通俗点讲，减少层级，flattenMerge和flattenConcat类似，但是可以设置并发数。

代码如下：

          val flow1 = (1..5).asFlow()
          val flow2 = flowOf("一","二","三","四","五")
            flowOf(flow1,flow2).flattenConcat().collect {
                Log.d(TAG.TAG,"flattenConcat $it")
            }
            flowOf(flow1,flow2).flattenMerge(2).collect {
                Log.d(TAG.TAG,"flattenMerge $it")
            }

日志如下：

2022-08-01 13:44:53.938 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 1
2022-08-01 13:44:53.938 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 2
2022-08-01 13:44:53.938 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 3
2022-08-01 13:44:53.938 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 4
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 5
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 一
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 二
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 三
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 四
2022-08-01 13:44:53.939 6409-6435/edu.test.demo D/Test-TAG: flattenConcat 五
2022-08-01 13:44:53.981 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 1
2022-08-01 13:44:53.981 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 一
2022-08-01 13:44:53.981 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 二
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 三
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 四
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 五
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 2
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 3
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 4
2022-08-01 13:44:53.982 6409-6435/edu.test.demo D/Test-TAG: flattenMerge 5

分析：

• 是不是看起来区别不大，都是将两层变成了异常，即展开、铺平，flattenConcat 是按顺序执行展平的，flattenMerge 设置参数为2之后则两个是并发执行的，可以看到结果先打印了一个1，再是大写一二三四五，然后是2345，当然这也不是一成不变的，也可能是其他的，也可能按顺序，因为是并发执行的。

5. flatMapConcat、flatMapMerge、flatMapLatest，这三个看起来比较类似，却别在于，flatMapConcat将元素转换成flow然后展平，flatMapMerge则在flatMapConcat的基础上可以设置并发数，则类似于collectLatest，展平最新的值。

代码如下：

            (1..5).asFlow().flatMapConcat {
                flowOf(it)
            }.collect {
                Log.d(TAG.TAG, "flatMapConcat $it")
            }
            (1..5).asFlow().flatMapMerge(5) {
                flowOf(it)
            }.collect {
                Log.d(TAG.TAG, "flatMapMerge $it")
            }
            (1..5).asFlow().flatMapLatest {
                delay(100)
                flowOf(it)
            }.collect {
                Log.d(TAG.TAG, "flatMapLatest $it")
            }

日志如下：

2022-08-01 14:00:36.995 7281-7307/edu.test.demo D/Test-TAG: flatMapConcat 1
2022-08-01 14:00:36.995 7281-7307/edu.test.demo D/Test-TAG: flatMapConcat 2
2022-08-01 14:00:36.995 7281-7307/edu.test.demo D/Test-TAG: flatMapConcat 3
2022-08-01 14:00:36.995 7281-7307/edu.test.demo D/Test-TAG: flatMapConcat 4
2022-08-01 14:00:36.995 7281-7307/edu.test.demo D/Test-TAG: flatMapConcat 5
2022-08-01 14:00:37.036 7281-7307/edu.test.demo D/Test-TAG: flatMapMerge 1
2022-08-01 14:00:37.036 7281-7307/edu.test.demo D/Test-TAG: flatMapMerge 2
2022-08-01 14:00:37.036 7281-7307/edu.test.demo D/Test-TAG: flatMapMerge 4
2022-08-01 14:00:37.036 7281-7307/edu.test.demo D/Test-TAG: flatMapMerge 5
2022-08-01 14:00:37.038 7281-7307/edu.test.demo D/Test-TAG: flatMapMerge 3
2022-08-01 14:00:37.161 7281-7307/edu.test.demo D/Test-TAG: flatMapLatest 5

分析：

• 可以看出flatMapConcat是按顺序进行map+展平操作。
• flatMapMerge设置并发数之后顺序不确定。
• flatMapLatest因为在闭包内设置了延时，所以只收集到了最新的值5。

回调操作符

1. onStart、onCompletion 这个很好理解，开始和结束。

代码如下：

            (1..5).asFlow().onStart {
                Log.d(TAG.TAG,"onStart")
            }.onCompletion {
                Log.d(TAG.TAG,"onCompletion")
            }.collect {
                Log.d(TAG.TAG,"collect is $it")
            }

日志如下：

2022-08-01 14:11:46.202 7384-7411/edu.test.demo D/Test-TAG: onStart
2022-08-01 14:11:46.203 7384-7411/edu.test.demo D/Test-TAG: collect is 1
2022-08-01 14:11:46.203 7384-7411/edu.test.demo D/Test-TAG: collect is 2
2022-08-01 14:11:46.203 7384-7411/edu.test.demo D/Test-TAG: collect is 3
2022-08-01 14:11:46.203 7384-7411/edu.test.demo D/Test-TAG: collect is 4
2022-08-01 14:11:46.204 7384-7411/edu.test.demo D/Test-TAG: collect is 5
2022-08-01 14:11:46.204 7384-7411/edu.test.demo D/Test-TAG: onCompletion

2. catch 处理异常，也可以重新emit值。

代码如下:

          flow {
                emit(1)
                throw IllegalArgumentException("exception throw")
                emit(2)
            }.catch {
                emit(-1)
                Log.e(TAG.TAG,"exception is $it")
            }.collect {
                Log.d(TAG.TAG,"collect is $it")
            }

日志如下：

2022-08-01 14:17:20.843 7465-7491/edu.test.demo D/Test-TAG: collect is 1
2022-08-01 14:17:20.844 7465-7491/edu.test.demo D/Test-TAG: collect is -1
2022-08-01 14:17:20.844 7465-7491/edu.test.demo E/Test-TAG: exception is java.lang.IllegalArgumentException: exception throw

分析：

• 可以看出catch捕获到了我们在flow中抛出的IllegalArgumentException异常，并且重新发送了特定值-1也被collect接收到了。
• emit(2)在抛出异常之后没有继续执行。

3. onEmpty 在flow为空的时候会回调

代码如下：

             flow<Int> {
            }.onEmpty {
                Log.d(TAG.TAG, "onEmpty")
            }.collect {
                Log.d(TAG.TAG, "collect is $it")
            }

日志如下：

2022-08-01 14:22:01.544 7598-7625/edu.test.demo D/Test-TAG: onEmpty

4. onEach 是在每次collect之前调用。

代码如下：

            flow<Int> {
                emit(1)
                emit(2)
                emit(3)
            }.onEach {
                Log.d(TAG.TAG, "onEach $it")
            }.collect {
                Log.d(TAG.TAG, "collect is $it")
            }

日志如下：

2022-08-01 14:26:11.313 8010-8035/edu.test.demo D/Test-TAG: onEach 1
2022-08-01 14:26:11.313 8010-8035/edu.test.demo D/Test-TAG: collect is 1
2022-08-01 14:26:11.313 8010-8035/edu.test.demo D/Test-TAG: onEach 2
2022-08-01 14:26:11.313 8010-8035/edu.test.demo D/Test-TAG: collect is 2
2022-08-01 14:26:11.313 8010-8035/edu.test.demo D/Test-TAG: onEach 3
2022-08-01 14:26:11.314 8010-8035/edu.test.demo D/Test-TAG: collect is 3

总结

• 本篇主要介绍了两个部分操作符，分别为合并操作符和回调操作符。
• 合并操作符，主要实现多个flow的一些合并、展平等操作；回调操作符，主要在flow执行的特定阶段回调展示或处理操作。
• 本篇为自己学习及使用过程中的总结，难免存在错误或思维局限，欢迎大家讨论指正。