目录
深入学习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
