前言

这篇文章主要介绍Flow的一些基础使用方法
同时介绍如何用Flow请求网络数据
下面开始！

什么是Flow

Flow翻译过来，是“流”的意思
举例说明，在大自然中，常见的如水流
是从高往低流动的
那么在计算机世界里，所谓的“流”
其实指的是数据流
也就是从获取原始数据，到进行处理，最后使用的过程
比如拿到一个json，转换为bean
然后进行筛选和过滤
拿到最后要用的最终数据
这个过程就称之为数据的流动
如下图：

而为了处理这个过程
我们就可以使用到Flow这个工具

Flow流的使用

简单使用

要使用Flow，首先需要导入协程相关工具类：

    implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.7-mpp-dev-11'implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.7-mpp-dev-11'

最简单的Flow使用：

suspend fun flow1() {flow<Int> {(0..4).forEach {emit(it)//生产者发送数据}}.collect {Log.d("flow1", "it:$it")//消费者处理数据}
}

我们分析下这段代码：
1：（0…4）是一个0-4的列表，这个是原始数据
2：emit其实就是把原始数据发送出去
3：collect用来收集发送的原始数据，并且内部自己打印了收到的数据
4：flow { } 函数包裹的代码块就是负责发送数据的，这个函数返回一个Flow对象
注意：Flow流是“冷流”
意思是 collect 被调用后 flow 内的方法体才会被调用
如果没有collect方法，不管如何emit，数据都是不会发送出去的

流操作符

Flow对于数据的操作，为我们提供了一系列的API
我们称之为“操作符”
一般分为“流构建器”、“中间操作符”和“末端操作符”
流构建器一般用来构建Flow流对象
中间操作符仅仅只是预先定义一些对流的操作方式，
比如过滤，转换等
并不会主动触发动作执行
而末端操作符则是对流的最终处理
比如collect就是末端操作符
下面就介绍一些操作符

流构建器

flowof

可以将 flowOf 内的可变长参数一一发射

flowOf(1, 2, 5, 4).collect {println(it)
}

asFlow

flowOf 可以将集合转换成 flow 发射

suspend fun asFlowM(){listOf(1,2,9,0,8).asFlow().collect{println(it)}
}

中间操作符

map

我们可以再 map 中执行一些过渡操作，
比如本例中将生产者发送的数据*9，然后再发射给消费者
值得一提的是，我们是可以再 map 中进行异步操作的
注意，这个map和集合没什么关系，别被误导了

suspend fun mapM(){(1..9).asFlow().map {it*9}.collect{println(it)}
}

transform

transform 主要强调的是类型的转换

(1..3).asFlow() // 一个请求流//transform中的泛型<Int,String> 表示将Int类型转换为String后，继续发射.transform<Int, String> { request ->emit("transform Int to String $request")}.collect { response -> println(response) }

take

限长操作符 take 可以限定我们要消费的数据的数量，见代码

(1..9).asFlow().take(3).collect {println(it)
}

conflate

当生产者发射数据速度大于消费者的时候，消费者只能拿到生产者最新发射的数据

suspend fun conflate(){flow<Int> {(1..9).forEach {delay(100)emit(it)}}.conflate().collect {delay(300)println(it)}
}

比如上面这段代码，因为有conflate的存在，输出如下：

如果没有conflate存在输出如下：

两者对比，明显能发现使用conflate的例子替我们忽略了很多无法即时处理的数据

collectLast

这个操作符的意思：如果生产者数据以及发射过来了，消费者还没有把上一个数据处理完，那么直接停止处理上一条数据，直接处理最新的数据


suspend fun collectLastM(){flow<Int> {(1..9).forEach {delay(100)emit(it)}}.collectLatest {delay(800)println(it)}
}

比如本例的输出为9

zip

zip操作符可以把两个流合并为一个流，然后再zip方法中将两个流发射的数据进行处理组合后继续发射给消费者，
如果两个流长度不一致，按比较短的流来处理：
1.两个流长度一致，都是3

suspend fun zipM(){val flow1 = (1..3).asFlow()val flow2 = flowOf("李白","杜甫","安安安安卓")flow1.zip(flow2){a,b->"$a : $b"}.collect {println(it)}
}

输出：

1 : 李白
2 : 杜甫
3 : 安安安安卓

上面的代码我们进行一下改变，将flow1的长度改为5

val flow1 = (1..5).asFlow()

查看输出结果：

1 : 李白
2 : 杜甫
3 : 安安安安卓

所以验证一下我们开头的结论，两个长度不同的流zip合并，消费者输出的数据长度是较短的流的长度

combine

上一节zip的缺点我们清楚了，就是两个流长度不等的时候，较长的流后面部分无法输出

那么combine就是用来解决zip这个缺点的（也很难说是缺点，只是应用场景不同罢了，你姑且可以认为是缺点）


suspend fun combineM(){val flowA = (1..5).asFlow()val flowB = flowOf("李白","杜甫","安安安安卓")flowA.combine(flowB){a,b->"$a : $b"}.collect {println(it)}
}

输出日志：

1 : 李白
2 : 李白
2 : 杜甫
3 : 杜甫
3 : 安安安安卓
4 : 安安安安卓
5 : 安安安安卓

我们的两个流，数字流长度为5，字符串流为3。

实现的效果简单逻辑分析：

flow发射1，flow2发射 ”李白“ ，打印：1 : 李白
flow发射2，flow2未发射数据  ，打印：2 : 李白
flow未发射，flow2发射 ”杜甫“ ，2 : 杜甫
flow发射3，flow2未发射 ，打印：3 : 杜甫
flow未发射，flow2发射 ”安安安安卓“ ，打印：3 : 安安安安卓
flow发射4，flow2发射完成  ，打印：4 : 安安安安卓
flow发射5，flow2发射完成  ，打印：5 : 安安安安卓

onCompletion

使用onCompletion可以再流完成的时候再发送一个值

 flowOf(1, 23, 5, 3, 4).onCompletion {println("流操作完成")emit(12344)//这里不返回值也没关系}.collect {println(it)}

输出：

1
23
5
3
4
流操作完成
12344

末端操作符

toList

会把数据消费到一个 List 列表中

suspend fun toList():List<Int> {return (1..9).asFlow().filter { it % 2 == 0 }.toList()
}

toSet

同 toList

frist

获取第一个元素

suspend fun firstM(): Int {return (2..9).asFlow().filter { it % 2 == 1 }.first()
}

reduce

reduce 的兰布达表达式会提供运算公式负责计算。

在 reduce 的兰布达表达式中，可以对当前要消费的值和之前计算的值进行计算，得到新值返回。所有值消费完成后返回最终值

suspend fun reduceM():Int {return (1..9).asFlow().reduce { accumulator, value ->println("$accumulator : $value")accumulator + value}
}

buffer

buffer可以缓存生产者数据，不会被消费者阻塞

suspend fun bufferM() {val startMillis = System.currentTimeMillis()flow<Int> {(1..3).forEach {delay(300)emit(it)}}.buffer(4).collect {delay(400)println(it)println("时间已经过了${System.currentTimeMillis() - startMillis}")}
}

代码执行打印日志：

1
时间已经过了745
2
时间已经过了1148
3
时间已经过了1552

如果我们没有用buffer，那么总时长应该2100ms
使用了buffer总时长是：1552=300+400*3
所以使用buffer的时候生产者可以并发发射数据，不会被消费者阻塞

流异常

使用try/catch包裹流
我们是可以使用try/catch来收集流异常的，但是不建议用这种方法
使用flow的catch操作符处理流
使用flow 的catch操作符处理异常更优雅
不过catch也有缺点，它只能捕获生产者的异常不能捕获消费者的异常


suspend fun trycatch() {flow<Int> {(1..3).forEach {if (it == 2) {//故意抛出一个异常throw NullPointerException("强行空指针，嘿嘿嘿嘿")}emit(it)}}.catch {e->e.printStackTrace()emit(-1)//异常的情况下发射一个-1}.collect{println(it)}
}

消费者的异常如何处理
尝试在消费者中抛出异常，查看是否可以被捕获

 flow<Int> {for (i in 1..3) {emit(i)}}.catch {emit(-1)}.collect {if(it==2){//在消费者中抛出数据throw IllegalArgumentException("数据不合法")}println(it)}

输出：

1
Exception in thread "main" java.lang.IllegalArgumentException: 数据不合法at HahaKt$consumerCatch$$inlined$collect$1.emit(Collect.kt:138)

将异常代码放在onEach中catch异常

suspend fun consumerCatch() {flow<Int> {for (i in 1..3) {emit(i)}}.onEach {if (it == 2) {//与上面的不同，在消费之前先用onEach处理一下throw IllegalArgumentException("数据不合法")}}.catch {emit(-1)}.collect {println(it)}
}

输出：

1
-1

Flow请求网络数据

上面介绍了Flow对数据的处理方法
我们可以配合Retrofit在我们的项目中进行使用
比如我们先定义一个请求，返回类型为Flow数据流

    @POST(ServerName.BS_API + "test/url")fun getMessage(): Flow<List<Bean>>

使用viewModelScope和viewmodel的生命周期绑定
然后对返回的数据进行过滤，得到最终需要的数据：

        viewModelScope.launch {repository.getMessage()?.filter {it.isNotEmpty()}?.collect {//得到数据}}

这样，我们就可以用Flow对返回回来的数据做各种复杂的处理了
当然，这些还都是基本用法，可以自己根据实际业务情况进行更多的封装

（原创）Flow数据流的使用

前言