通过 Flow，我们可以以响应式的编程方式进行协程代码的编写。Flow 类似于协程版本的 RxJava，但是比起 RxJava，它会更加简单，更加容易上手。

基本使用

        GlobalScope.launch(Dispatchers.Main) {flow {repeat(3) { // 重复3次Thread.sleep(1000) //模拟耗时计算工作emit(it) // 发送结果}}.flowOn(Dispatchers.Default) // 切换到工作子线程.collect {// 运行在调用 collect 的线程当中，在此 case 中是 Dispatchers.Main，也就是主线程。Log.i("TAG", "it: $it") //打印 0 1 2}}

以上面的代码为例，我们使用 flow 主要有以下几步：
1.定义生产者函数，在函数中将结果通过 emit 函数发出。
2.如果需要指定生产者函数所运行的线程，通过 flowOn 函数指定并传入一个 CoroutineContext。
3.调用 collect 函数收集生产者函数的计算结果，collect 是一个 suspend 函数，运行在调用 collect 函数的线程当中，且在收集完成之前会一直挂起，因此需要在挂起函数中被调用。

操作符

flow 提供了非常丰富且便捷的操作符。

flowOn

用于指定生产者函数运行在哪个 CoroutineContext，如第一个例子所示。

onEach

每当生产者函数发送结果时，都会被调用。

    flow {repeat(3) {emit(it)}}.onEach {println("onEach $it") // 打印 0 1 2}.collect {println("collect $it") // 打印 0 1 2}

launchIn

指定在哪个 CoroutineContext 中进行收集，可以从源码中看出它的实现。

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

通过 launchIn，我们可以将第一个例子改成如下代码：

        GlobalScope.launch(Dispatchers.Main) {flow {repeat(3) { // 重复3次Thread.sleep(1000) //模拟耗时计算工作emit(it) // 发送结果}}.flowOn(Dispatchers.Default) // 切换到工作子线程.onEach {// 运行在调用 launchIn 的线程当中，在此 case 中是 Dispatchers.Main，也就是主线程。Log.i("TAG", "it: $it") //打印 0 1 2}.launchIn(this)}

onEmpty

当生产者函数没有发送任何结果时会被调用。

    flow { }.onEmpty { emit(-1) }.collect { println("testOnEmpty: $it") } // 打印 -1flow { emit(1) }.onEmpty { emit(-1) }.collect { println("testOnEmpty: $it") } // 打印 1

onStart

在收集开始之前调用

    // 打印 start 0 1 2flow {repeat(3) {emit(it)}}.onStart { println("start") }.collect { println("$it") }

onCompletion

在收集完成后调用

    // 打印 0 1 2 completion flow {repeat(3) {emit(it)}}.onCompletion { println("completion") }.collect { println("$it") } 

catch

当上游操作符或者生产者函数出现异常时被调用：

        // 打印// collect: 0// catch: java.lang.NullPointerExceptionflow {repeat(3) {if (it == 1) {throw NullPointerException()}emit(it)}}.catch { println("catch: $it") }.collect { println("collect: $it") }

map

将上游的结果进行变换后，发送给下游

        // 打印 0 1 4 9 16flow {repeat(5) {emit(it)}}.map { it * it }.collect { println("$it") }

flatMapConcat

根据上游 flow 的值生成新的 flow，再将新的 flow 的结果发送给下游收集者，下面是打印99乘法表的例子

        //打印//1 2 3 4 5 6 7 8 9//2 4 6 8 10 12 14 16 18//3 6 9 12 15 18 21 24 27//4 8 12 16 20 24 28 32 36//5 10 15 20 25 30 35 40 45//6 12 18 24 30 36 42 48 54//7 14 21 28 35 42 49 56 63//8 16 24 32 40 48 56 64 72//9 18 27 36 45 54 63 72 81flow {for (i in 1..9) {emit(i)}}.flatMapConcat {flow {for (j in 1..9) {emit(it * j)}println()}}.collect { print("$it ") }

collectLatest

当收集函数被挂起时，如果生产者函数发送了新结果，则取消当前被挂起的收集函数，并基于新结果重新运行收集函数。换句话说就是确保只接收最新的数据。

        //打印//0 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//1 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//2 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//3 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//4 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//5 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//6 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//7 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//8 kotlinx.coroutines.flow.internal.ChildCancelledException: Child of the scoped flow was cancelled//9flow {repeat(10) {delay(100)emit(it)}}.collectLatest {try {delay(200)} catch (e: CancellationException) {println("$it $e")throw e}println("$it") //到这才算收集成功}

由于 collect 函数每次收集时都会挂起 200 毫秒，而生产函数每 100 毫秒就会发送一次数据，因此 collect 函数每次的挂起都会因为有新结果而被取消，直到最后一个数据发送时才能成功收集。