Kotlin Flow 一 Flow 的创建和使用

Flow 和 RxJava 差不多，不过 Flow 是和协程一起使用的 API。

简单的例子

flow {
    emit(1)
    emit(2)
}.collect {
    println(it)
}

在 Flow 中可以使用 emit 发送数据，相当于 RxJava 中的 onNext() 。

创建 Flow

Flow 可以使用函数，集合，数组等来进行创建。

• flowOf()

  flowOf(1, 2, 3, 4, 5)
      .onEach {
          delay(it * 100L)
      }
      .collect {
          println(it)
      }
  
  

• asFlow

  listOf(1, 2, 3).asFlow()
      .collect {
          println(it)
      }
  arrayOf('a', 'b', 'c')
          .asFlow()
          .collect {
              println(it)
          }
  
  

  asFlow 可以将任何使用 Iterator 和数组转换为 Flow。

  {
     1 + 1
  }
  .asFlow()
      .collect {
          println(it)
      }
  suspend {
      delay(1000)
      1 + 1
  }
      .asFlow()
      .collect { 
          println(it)
      }
  
  

  asFlow 还可以将函数（挂起函数也可以）的结果转换为 flow 。

• channelFlow

  channelFlow {
      for (i in 1..5) {
          delay(i * 100L)
          send(i)
      }
  }.collect {
      println(it)
  }
  
  

  channelFlow 是通过协程中的 channel 来实现的，所以它的发送数据和普通的 flow 不太一样。 channelFlow 是通过 send 来发送的。

  flow 是冷的流。在没有切换协程的情况下，生产者和消费者是同步非阻塞的。

  channel 是热流。 channelFlow 实现了生产者和消费者的异步非阻塞模型。

切换协程

Flow 是专用于 Kotlin 的 API，所以在会对协程进行切换，而不是协程。

flow {
    for (i in 1..5) {
        delay(100L)
        emit(i)
    }
}
    .map {
        it * it
    }
    .flowOn(Dispatchers.IO)
    .collect {
        println(it)
    }

相比于 RxJava，flow 切换协程更加简单，只需要调用 flowOn 即可切换协程。

Flow 的取消

如果 Flow 是在一个挂起函数中被挂起了，那么 Flow 是可以取消的，否则不可以取消。

Terminal flow operator

Flow 的 API 和 Java 的相似，都有 Intermediate Operations、Terminal Operations。

Flow 的 Terminal 运算符可以是 suspend 函数，如 collect、single、reduce、toList 等，也可以是 launchIn 运算符，用户在指定的 CoroutineScope 内使用 flow。

整理一下 Flow 的 terminal operator，如下：

• collect
• single/first
• toList/toSet/toCollection
• count
• fold/reduce
• launchIn/produceIn/broadcastIn

collect

接受 flow 下数据流：

flow {
    emit(1)
    emit(2)
}.collect {
    println(it)
}

single

如果 flow 中只发送一个数据可以通过 single 来返回，但是如果多于一个数据就会抛出 java.lang.IllegalArgumentException: Flow has more than one element。

println("------flow terminal single")
val singleValue = flowOf(1)
    .single()
println("single value = $singleValue")

//  flowOf(1, 2)
//      .single()

single 的源码如下：

public suspend fun <T> Flow<T>.single(): T {
    var result: Any? = NULL
    collect { value ->
        require(result === NULL) { "Flow has more than one element" }
        result = value
    }

    if (result === NULL) throw NoSuchElementException("Flow is empty")
    return result as T
}

first

获取 Flow 中的第一个元素。

val firstValue = flowOf(1, 2, 3)
    .first()
println("first value = $firstValue")

first 的源码如下：

public suspend fun <T> Flow<T>.first(): T {
    var result: Any? = NULL
    collectWhile {
        result = it
        false
    }
    if (result === NULL) throw NoSuchElementException("Expected at least one element")
    return result as T
}

toList

toList 把 flow 中的数据添加到集合中。

val list = flowOf(1, 2, 3)
    .toList()
println("list = $list")
val myList = mutableListOf(100, 200)
val myList2 = flowOf(1, 2, 3)
    .toList(myList)
println("add myList as param res = $myList2")

toList 源码如下：

public suspend fun <T> Flow<T>.toList(destination: MutableList<T> = ArrayList()): List<T> = toCollection(destination)
public suspend fun <T, C : MutableCollection<in T>> Flow<T>.toCollection(destination: C): C {
    collect { value ->
        destination.add(value)
    }
    return destination
}

fold

用初始值作为初累计结果，然后再将累计的结果和每一个值做 operator 运算，然后将结果作为累计，用于下一次的计算。

Accumulate value starting with initial value and applying operation current accumulator value and each element.

val sum = flowOf(1, 2, 3)
    .fold(0) { acc, value ->
        acc + value
    }

println("use flow fold to sum, result = $sum")

fold 源码如下：

public suspend inline fun <T, R> Flow<T>.fold(
    initial: R,
    crossinline operation: suspend (acc: R, value: T) -> R
): R {
    var accumulator = initial
    collect { value ->
        accumulator = operation(accumulator, value)
    }
    return accumulator
}

reduce

reduce 和 fold 相似，不同的是 reduce 没有初始值。

val sum = flowOf(1, 2, 3)
    .reduce { accumulator, value ->
        accumulator + value
    }
println("use flow reduce to sum, result = $sum")

reduce 源码如下：

public suspend fun <S, T : S> Flow<T>.reduce(operation: suspend (accumulator: S, value: T) -> S): S {
    var accumulator: Any? = NULL

    collect { value ->
        accumulator = if (accumulator !== NULL) {
            @Suppress("UNCHECKED_CAST")
            operation(accumulator as S, value)
        } else {
            value
        }
    }

    if (accumulator === NULL) throw NoSuchElementException("Empty flow can't be reduced")
    @Suppress("UNCHECKED_CAST")
    return accumulator as S
}

launchIn

launchIn 会在执行的协程内调用 Flow 的 collect() 方法。

val simpleFlow: Flow<Int> = flow {
    for (i in 1..10) {
        emit(i)
        println("launchIn ${currentCoroutineContext()[CoroutineName]?.name} emit $i")
        delay(100)
    }
}
coroutineScope {
    launch(CoroutineName("SimpleCoroutine")) {
        simpleFlow.launchIn(this)
    }
}

launchIn 的源码如下：

public fun <T> Flow<T>.launchIn(scope: CoroutineScope): Job = scope.launch {
    collect() // tail-call
}