机票直达

Kotlin 协程之 Flow 详解一 冷流的生产和消费
Kotlin 协程之 Flow 详解二 常用操作符
Kotlin 协程之 Flow 详解三 还是常用操作符

概述

Flow类似RxJava，当然也有许多操作符，不过Flow的操作符学习要比RxJava的操作符简单直白一些，本人感觉更喜欢Flow一些，本文将介绍一部分中间操作符的使用（合并、回调、线程调度）

合并操作符（属于中间操作符）

• zip（常用）

对两个流进行组合，分别从二者取值，一旦一个流结束了，那整个过程就结束了

   val flow1  = listOf(0,1,2,3,4,5).asFlow() // zip后， 5将被丢弃
   val flow2 = listOf(6,7,8,9,10).asFlow()
   flow1.zip(flow2){value1,value2->
       "it is $value1,$value2"
   }.collect {
      println(it) //结果 it is 06 、17、 28、 39、410
   }

• merge（常用）

是将两个流合并起来，将每个值依次发出来

   val flow1  = listOf(0,1,2,3,4,5).asFlow()
   val flow2 = listOf(6,7,8,9,10).asFlow()
   merge(flow1,flow2).collect {
      println(it) //结果 0,1,2,3,4,5,6,7,8,9,10
   }

• combine、combineTransform

combine 组合每个流最新发出的值，combineTransform = combine+Transform

val flow1 = flowOf(1, 2).onEach { delay(100) }
val flow2 = flowOf("3", "4", "5").onEach { delay(150) }
flow1.combine(flow2) { value1, value2 ->
 "$value1,$value2"
 }.collect {
  println(it)  // 结果 13  23  24 25"
}

• flattenConcat（常用）

flattenConcat 展平多个流，以顺序方式将给定的流展开为单个流

flow {
    emit(flowOf(1, 2, 3,4,5))
    emit(flowOf(6,7,8,9,10))
 } .flattenConcat().collect {
     println(it) //结果 1,2,3,4,5,6,7,8,9,10
 }

• flattenMerge

和 flattenConcat 大抵一样，区别是可以设置并发数

flow {
    emit(flowOf(1, 2, 3,4,5))
    emit(flowOf(6,7,8,9,10))
 } .flattenMerge(2).collect {
     println(it) //结果 1,2,3,4,5,6,7,8,9,10 ，顺序不固定，有并发
 }

• flatMapLatest

和其他Lastest操作符一样，来不及转换时就取消

• flatMapContact

flattenConcat + map

回调操作符

• onStart、onCompletion（常用）

开始和结束

flow {
    emit(1)
    emit("2")
    emit("3")
}.onStart {
    println("onStart")
}.onCompletion {
    println("onCompletion")
}.collect {
    println(it)
}
// 结果 onStart 1 2 3 onCompletion

*onEmpty

在flow为空的时候会回调，很好理解，不再示例

• onEach

是在每次collect之前调用，很好理解，不再示例

线程调度操作符

• flowOn（常用）

用于将上游的流切换到指定协程上下文的调度器中执行，同时不会把协程上下文暴露给下游的流，即flowOn方法中协程上下文的调度器不会对下游的流生效

launch(Dispatchers.Main) {
  flow {
    emit(1) // IO线程
    emit(2) // IO线程
    emit(3) // IO线程
  }.flowOn(Dispatchers.IO).map {
    "it is $it" // Default线程
  }.flowOn(Dispatchers.Default).collect {
    println(it) //主线程
  }
}

功能性操作符

• catch （常用）

异常处理，也可以重新发送一个值

flow {
    emit(1)
    throw IllegalArgumentException("exception")
    emit("3")
}.onStart {
    println("onStart")
}.catch{
    println("exception")
    emit(2)
}.onCompletion {
    println("onCompletion")
}.collect {
    println(it)
}
// 结果 onStart 1 exception 2  onCompletion
// emit("3") 抛出异常以后 没有继续发送

• retryWhen

有条件的进行重试 ，lambda 中有两个参数: 一个是 异常原因，一个是当前重试的 index (从0开始)。
lambda 的返回值 为 Boolean ，true则继续重试 ,false 则结束重试。

flow<Int> {
    throw IOException("error")
 } .retryWhen {cause,attempt->
     if(attempt > 3){
        return@retryWhen false
     }
    cause is IOException
 }
//结果将retry 3次