深入学习Kotlin之Flow(二),Flow的操作符,协程的背

目录

深入学习Kotlin之Flow(一),什么是Flow?Flow的基本使用)
深入学习Kotlin之Flow(二),Flow的操作符,以及协程的背压

类似集合的函数是Api，Flow中也有许多操作符,常见的有

• map
• filter
• 末端操作符(collect就是一种末端操作符)
• flowOn
• retry
• zip
• Combine
• 协程背压(buffer,conflate,collectLatest)

这里我们简单列表一些常用的操作符的例子:

(1)map操作符

使用map我们可以将最终结果映射为其他类型，融合了Rxjava的map与flatMap的功能
代码如下所示：

fun changeData(value: Int): String {
    return "打印的结果是：${value}"
}

fun main() {
    runBlocking {
        loadData1().map {
            changeData(it)
        }.collect{
            println(it)
        }
    }

}

我们通过map操作符将结果映射为字符串的形式，运行结果

打印的结果是：1
打印的结果是：2
打印的结果是：3

(2)filter操作符

通过filter 我们可以对结果集添加过滤条件，如下所示，我们仅打印出大于1的值

 runBlocking {
        loadData1().filter {
            it > 1
        }.collect {
            println(it)
        }
    }

运行结果:

2
3

所有的操作符都是可以一起使用的，并非只能单独使用

(3) 末端操作符

我们上面调用的collect是末端操作符，在Flow中除了collect之外 还有toList、reduce、fold,onEach等操作符。

toList操作符我们可以很明显的知道意为转换为list集合，而reduce 和 fold 则可将最终的值转为单一的值

fun main() {
    runBlocking {
        var data = loadData1().reduce { a, b ->
            a + b
        }
        println(data)
    }
}

如上代码，我们将Flow的每个结果最终求和
运行结果

6

(4) flowOn操作符

Flow的代码块是执行在执行时的上下文中比如 我们不能通过在flow中指定线程来运行Flow代码中的代码

如下所示：

fun loadData1() = flow {
    withContext(Dispatchers.Default){
        for (i in 1..3) {
            delay(1000)
            emit(i)
        }
    }

}


fun main() {
    runBlocking {
        loadData1().collect { value -> println("Collected $value") }
    }
}

此种运行方式，将会抛出异常

Exception in thread "main" java.lang.IllegalStateException: Module with the Main dispatcher had failed to initialize. For tests Dispatchers.setMain from kotlinx-coroutines-test module can be used
at kotlinx.coroutines.internal.MissingMainCoroutineDispatcher.missing(MainDispatchers.kt:113)

核心就是切换线程 类似Rxjava的subscribeOn(Schedulers.io())
那么我们如何指定Flow代码块中的上下文呢，我们需要使用flowOn操作符,我们将Flow代码块中的代码指定在IO线程中，代码如下所示：

fun loadData1() = flow {
    for (i in 1..3) {
        delay(1000)
        emit(i)
    }
}.flowOn(Dispatchers.IO)

这样我们就把Flow代码块中的事情放到了IO线程中

(5) retry操作符

有异常的情况下重试

// 5秒轮询一次 错误重试三次    
  suspend fun flowDemo(): LiveData<String> {
        return flow {
            while (true) {
                emit(repository.sendNetworkRequestSuspend())
                delay(5000)
            }
        }.map {
            it.html_url
        }.retry(3).catch {
            // 类似于RxJava的onError
            Log.e(TAG, it.message)
        }.onCompletion {
            // 类似于Rxjava中的onComplete
            Log.i(TAG, "finally")
        }.flowOn(Dispatchers.IO).asLiveData()
    }
  
 val currentName = liveData {
        try {
            emitSource(flowDemo())
        } catch (e: Throwable) {
            e.printStackTrace()
        }
    }

Retrywhen：满足条件为true时重试

(6) zip操作符

合并两个flow数据流，会分别对两个流合并处理，也就是快的流要等慢的流发射完才能合并。一般用作合并两个网络请求返回数据

val flow = flowOf（1、2、3）.onEach {delay（10）}
val flow2 = flowOf（“ a”，“ b”，“ c”，“ d”）。onEach {delay（15）}
flow.zip（flow2）{i，s-> i.toString（）+ s} .collect {
    println（it）
}

运行结果:

1a
2b
3c

(7) combine操作符

使用 combine 合并时，每次从 flowA 发出新的 item ，会将其与 flowB 的最新的 item 合并。

fun main() = runBlocking {
    val flowA = (1..5).asFlow().onEach { delay(100)  }
    val flowB = flowOf("one", "two", "three","four","five").onEach { delay(200)  }
    flowA.combine(flowB) { a, b -> "$a and $b" }
        .collect { println(it) }
}

运行结果:

1 and one
2 and one
3 and one
3 and two
4 and two
5 and two
5 and three
5 and four
5 and five

(8) 协程背压(buffer,conflate,collectLatest)

Kotlin协程支持背压。Kotlin流程设计中的所有函数都标有suspend修饰符-具有在不阻塞线程的情况下挂起调用程序执行的强大功能。因此，当流的收集器不堪重负时，它可以简单地挂起发射器，并在准备好接受更多元素时稍后将其恢复。

buffer操作符

buffer() 对应RxJava中的 BUFFER 策略

没有固定大小，可以无限制添加数据，不会抛出 MissingBackpressureException 异常，但可能会导致 OOM

fun main() = runBlocking {
    var start = 0L
    val time = measureTimeMillis {
        (1..5)
                .asFlow()
                .onStart { start = System.currentTimeMillis() }
                .onEach {
                    delay(100)
                    println("Emit $it (${System.currentTimeMillis() - start}ms) ")
                }
                .buffer()
                .flowOn(Dispatchers.IO)
                .collect {
                    println("Collect $it starts (${System.currentTimeMillis() - start}ms) ")
                    delay(500)
                    println("Collect $it ends (${System.currentTimeMillis() - start}ms) ")
                }
    }

    println("Cost $time ms")
}

运行结果

Emit 1 (109ms)
Collect 1 starts (115ms)

Emit 2 (219ms)
Emit 3 (324ms)
Emit 4 (426ms)
Emit 5 (531ms)
Collect 1 ends (618ms)
Collect 2 starts (618ms)
Collect 2 ends (1122ms)
Collect 3 starts (1123ms)
Collect 3 ends (1625ms)
Collect 4 starts (1625ms)
Collect 4 ends (2127ms)
Collect 5 starts (2127ms)
Collect 5 ends (2627ms)
Cost 2683 ms

conflate操作符

conflate() 对应 LATEST 策略,如果缓存池满了，新数据会覆盖老数据
将上面buffer()改成conflate()接口如下

Emit 1 (114ms)
Collect 1 starts (117ms)
Emit 2 (217ms)
Emit 3 (329ms)
Emit 4 (433ms)
Emit 5 (538ms)
Collect 1 ends (620ms)
Collect 5 starts (620ms)
Collect 5 ends (1124ms)
Cost 1171 ms

collectLatest()操作符

只处理最新的数据，这看上去似乎与 conflate 没有区别，其实区别大了：它并不会直接用新数据覆盖老数据，而是每一个都会被处理，只不过如果前一个还没被处理完后一个就来了的话，处理前一个数据的逻辑就会被取消。

flow {
  List(100) {
    emit(it)
  }
}.collectLatest { value ->
  println("Collecting $value")
  delay(100)
  println("$value collected")
}

运行结果

Collecting 0
Collecting 1
...
Collecting 97
Collecting 98
Collecting 99
▶ 100ms later
99 collected

(9) 其他操作符

• flattenMerge : flowA、flowB 作为单个流的执行。类似于Rxjava的的merge
• take：获取集合数据的前几个数据
• drop：过滤集合的前集几个数据
• onEach：遍历
• onStart：开始会执行事执行,在耗时操作的时候可以用来做loading

如果还想了解更多的Flow操作符号 参考官方文档-Kotlin-协程-Flow

(每天学习一点点.每天进步一点点,分享不宜路过点个赞呀,喜欢的点个关注后续更新不断)