Kotlin 学习笔记（五）—— Flow 数据流学习实践指北（一）（上）

开启掘金成长之旅！这是我参与「掘金日新计划 · 12 月更文挑战」的第1天，点击查看活动详情

最近马斯克收购了推特之后，马上就裁掉了 50% 的推特员工，这不禁让我想起了灭霸的响指... 还有苹果、亚马逊冻结招聘，英特尔、Lyft开启裁员计划，国内外都不好过啊，大家都开始勒紧裤腰带了···那么，我们打工人是不是也该刷刷题了···(笑Cry.jpg)

Kotlin 学习笔记艰难地来到了第五篇~ 在这一篇主要会说 Flow 的基本知识和实例。由于 Flow 内容较多，所以会分几个小节来讲解，这是第一小节，文章后面会结合一个实例介绍 Flow 在实际开发中的应用。

首先回想一下，在协程中处理某个操作，我们只能返回单个结果；而 Flow 可以按顺序返回多个结果，在官方文档中，Flow 被翻译为 数据流，这也说明了 Flow 适用于多值返回的场景。

Flow 是以协程为基础构建的，所以它可通过异步的方式处理一组数据，所要处理的数据类型必须相同，比如：Flow<Int>是处理整型数据的数据流。

Flow 一般包含三个部分：

1）提供方：负责生成数据并添加到 Flow 中，得益于协程，Flow 可以异步生成数据；

2）中介(可选)：可对 Flow 中的值进行操作、修改；也可修改 Flow 本身的一些属性，如所在线程等；

3）使用方：接收并使用 Flow 中的值。

提供方：生产者，使用方：消费者，典型的生产者消费者模式。

1. Flow 概述

Flow 是一个异步数据流，它可以顺序地发出数据，通过流上的一些中间操作得出结果；若出错可抛出异常。这些 “流上的中间操作” 包括但不限于 map、filter、take、zip 等等方法。这些中间操作是链式的，可以在后面再次添加其他操作方法，并且也不是挂起函数，它们只是构建了一条链式的操作并实时返回结果给后面的操作步骤。

流上的终端操作符要么是挂起函数，例如 collect、single、toList 等等，要么是在给定作用域内开始收集流的 launchIn 操作符。前半句好理解，后半句啥意思？这就得看一下 launchIn 这个终端操作符的作用了。它里面是这样的：

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

原来 launchIn 方法可以传入一个 CoroutineScope 协程作用域，然后在这个作用域里面调用 collect 方法。lifecycleScope、MainScope() 这些都是协程作用域，所以 launchIn  方法只不过是 scope.launch { flow.collect() } 的一种简写。

流的执行也被称之为收集流，并且是以挂起的方式，不是阻塞的。流最终的执行成功与否取决于流上的操作是否全部执行成功。collect 函数就是最常见的收集流函数。

1.1 冷流与热流

冷流（Cold Flow）：在数据被使用方订阅后，即调用 collect 方法之后，提供方才开始执行发送数据流的代码，通常是调用 emit 方法。即不消费，不生产，多次消费才会多次生产。使用方和提供方是一对一的关系。

热流（Hot Flow）：无论有无使用方，提供方都可以执行发送数据流的操作，提供方和使用方是一对多的关系。热流就是不管有无消费，都可生产。

SharedFlow 就是热流的一种，任何流也可以通过 stateIn 和 shareIn 操作转化为热流，或者通过 produceIn 操作将流转化为一个热通道也能达到目的。本篇只介绍冷流相关知识，热流会在后面小节讲解~

2. Flow 构建方法

Flow 的构造方法有如下几种：

1、 flowOf() 方法。用于快速创建流，类似于 listOf() 方法，下面是它的源码：

//code 2
public fun <T> flowOf(vararg elements: T): Flow<T> = flow {
    for (element in elements) {
        emit(element)
    }
}

所以用法也比较简单：

//code 3
val testFlow = flowOf(65,66,67)
lifecycleScope.launch {
    testFlow.collect {
        println("输出：$it")
    }
}
//打印结果：
//输出：65
//输出：66
//输出：67

注意到 Flow 初始化的时候跟其他对象一样，作用域在哪儿都可以，但 collect 收集的时候就需要放在协程里了，因为 collect 是个挂起函数。

2、asFlow() 方法。是集合的扩展方法，可将其他数据转换成 Flow，例如 Array 的扩展方法：

//code 4
public fun <T> Array<T>.asFlow(): Flow<T> = flow {
    forEach { value ->
        emit(value)
    }
}

不仅 Array 扩展了此方法，各种其他数据类型的数组都扩展了此方法。所以集合可以很方便地构造一个 Flow。

3、flow {···} 方法。这个方法可以在其内部顺序调用 emit 方法或 emitAll 方法从而构造一个顺序执行的 Flow。emit 是发射单个值；emitAll 是发射一个流，这两个方法分别类似于 list.add(item)、list.addAll(list2) 方法。flow {···} 方法的源码如下：

//code 5
public fun <T> flow(@BuilderInference block: suspend FlowCollector<T>.() -> Unit): Flow<T> = SafeFlow(block)

需要额外注意的是，flow 后面的 lambda 表达式是一个挂起函数，里面不能使用不同的 CoroutineContext 来调用 emit 方法去发射值。因此，在 flow{...} 中不要通过创建新协程或使用 withContext 代码块在另外的 CoroutineContext 中调用 emit 方法，否则会报错。如果确实有这种需求，可以使用 channelFlow 操作符。

//code 6
val testFlow = flow {
    emit(23)
//    withContext(Dispatchers.Main) { // error
//        emit(24)
//    }
    delay(3000)
    emitAll(flowOf(25,26))
}

4、channelFlow {···} 方法。这个方法就可以在内部使用不同的 CoroutineContext 来调用 send 方法去发射值，而且这种构造方法保证了线程安全也保证了上下文的一致性，源码如下：

//code 7
public fun <T> channelFlow(@BuilderInference block: suspend ProducerScope<T>.() -> Unit): Flow<T> =
    ChannelFlowBuilder(block)

一个简单的使用例子：

//code 8
val testFlow1 = channelFlow {
    send(20)
    withContext(Dispatchers.IO) { //可切换线程
        send(22)
    }
}
lifecycleScope.launch {
    testFlow1.collect {
        println("输出 = $it")
    }
}

5、MutableStateFlow 和 MutableSharedFlow 方法：都可以定义相应的构造函数去创建一个可以直接更新的热流。由于篇幅有限，有关热流的知识后面小节会再说明。

3. Flow 常用的操作符

Flow 的使用依赖于众多的操作符，这些操作符可以大致地分为 中间操作符 与 末端操作符 两大类。中间操作符是流上的中间操作，可以针对流上的数据做一些修改，是链式调用。中间操作符与末端操作符的区别是：中间操作符是用来执行一些操作，不会立即执行，返回值还是个 Flow；末端操作符就会触发流的执行，返回值不是 Flow。

一个完整的 Flow 是由 Flow 构建器、Flow 中间操作符、Flow 末端操作符 组成，如下示意图所示：

3.1 collect 末端操作符

最常见的当然是 collect 操作符。它是个挂起函数，需要在协程作用域中调用；并且它是一个末端操作符，末端操作符就是实际启动 Flow 执行的操作符，这一点跟 RxJava 中的 Observable 对象的执行很像。

熟悉 RxJava 的同学知道，在 RxJava 中，Observable 对象的执行开始时机是在被一个订阅者(subscriber) 订阅(subscribe) 的时候，即在 subscribe 方法调用之前，Observable 对象的主体是不会执行的。

Flow 也是相同的工作原理，Flow 在调用 collect 操作符收集流之前，Flow 构建器和中间操作符都不会执行。举个栗子：

//code 9
val testFlow2 = flow {
    println("++++ 开始")
    emit(40)
    println("++++ 发出了40")
    emit(50)
    println("++++ 发出了50")
}
lifecycleScope.launch {
    testFlow2.collect{
        println("++++ 收集 = $it")
    }
}
// 输出结果：
//com.example.myapplication I/System.out: ++++ 开始
//com.example.myapplication I/System.out: ++++ 收集 = 40
//com.example.myapplication I/System.out: ++++ 发出了40
//com.example.myapplication I/System.out: ++++ 收集 = 50
//com.example.myapplication I/System.out: ++++ 发出了50

从输出结果可以看出，每次到 collect 方法调用时，才会去执行 emit 方法，而在此之前，emit 方法是不会被调用的。这种 Flow 就是冷流。

3.2 reduce 末端操作符

reduce 也是一个末端操作符，它的作用就是将 Flow 中的数据两两组合接连进行处理，跟 Kotlin 集合中的 reduce 操作符作用相同。举个栗子：

//code 10
private fun reduceOperator() {
    val testFlow = listOf("w","i","f","i").asFlow()
    CoroutineScope(Dispatchers.Default).launch {
        val result = testFlow.reduce { accumulator, value ->
            println("+++accumulator = $accumulator  value = $value")
            "$accumulator$value"
        }
        println("+++final result = $result")
    }
}
//输出结果：
//com.example.myapplication I/System.out: +++accumulator = w  value = i
//com.example.myapplication I/System.out: +++accumulator = wi  value = f
//com.example.myapplication I/System.out: +++accumulator = wif  value = i
//com.example.myapplication I/System.out: +++final result = wifi

看结果就知道，reduce 操作符的处理逻辑了，两个值处理后得到的新值作为下一轮中的输入值之一，这就是两两接连进行处理的意思。

图1 中出现的 toList 操作符也是一种末端操作符，可以将 Flow 返回的多个值放进一个 List 中返回，返回的 List 也可以自己设置，比较简单，感兴趣的同学可自行动手试验。