kotlin flow 使用

1 创建flow

方式1

通过携程扩展函数FlowKt中的flow扩展函数可以直接构建flow，只需要传递FlowCollector收集器实现类就可以了

private fun create1(){val intFlow = createFlow()println("创建int flow: $intFlow")runBlocking {println("开始收集")intFlow.collect {println(it)}println("收集结束")}}/*** 创建flow 方式1*/private fun createFlow(): Flow<Int> = flow {delay(1000)emit(1)delay(1000)emit(2)delay(1000)emit(3)}

通过该flow扩展函数可以了解到如下信息
函数定义泛型时需要将泛型放到fun 和函数名中间
例如

public fun <T> flow(@kotlin.BuilderInference block: suspend kotlinx.coroutines.flow.FlowCollector<T>.() -> kotlin.Unit)

定义好之后，函数中传参也可以使用该泛型继续向下传递，例如这里传递给了FlowCollector接口，用于规定flow收集器应该发送什么数据给调用方

方式2

private fun create2() {// 创建flow 方式2println("创建flow 方式2")val flow2 = flowOf(1, 2, 3).onEach {delay(1000)}runBlocking {flow2.collect {println(it)}}}

flowof 扩展函数其实内部也是调用flow扩展函数，只不过flowof是将传递进来的可变参数，遍历了一遍，并且调用flow收集器的emit方法发送取出而已

方式3

 private fun create3() {println("创建flow 方式3")runBlocking {val flow3 = listOf(1,2,3).asFlow().onEach {delay(1000)}flow3.collect {println(it)}}}

调用list顶级接口类Iterable的asFlow方法，其实内部还是调用了flow扩展函数，将元素遍历之后emit出去的

2 取消或中断flow

   // 测试Flow的取消 取消Flow只需要取消它所在的携程即可private fun testCancelFlow() {runBlocking {withTimeoutOrNull(250) { // 在 250 毫秒后超时simple().collect { value -> println(value) }}println("Done")}}fun simple(): Flow<Int> = flow {for (i in 1..3) {delay(100)println("Emitting $i")emit(i)}}

通过withTimeoutOrNull 函数可以实现取消flow携程功能， 通过该功能可以处理类似于某些场景下的超时机制，兜底逻辑等，例如：
在30秒内处理不完数据状态的转换，就直接结束当前处理，只展示处理过后的数据就可以了。因为有些场景下，数据量较大，用户等待时间太长的情况下，一般用户无法接受，所以只能采用妥协的方式来规定一个最大等待时间，来处理该组数据。一般会用在对数据做某种优化处理，但这种处理即使处理不完的情况下，产品也能接受的情况下，才会采用该方案。否则不建议执行兜底方案处理。

3 Flow 操作符

collect 是Flow中最常用，最基础的末端操作符，主要用于Flow启动流的挂起函数。除了collect还有一些其他的操作符，例如：

• 转换为各种集合的操作符： toList/toSet/toCollection
• 用于流元素计算的操作符：reduce / fold
• 计算数量的操作符：count
• 用于切换协程作用域线程的操作符：launcherIn/productIn/broadcastIn, 这里的productIn以及broadcastIn 暂时没有了解清楚其用法和作用，暂时放在这里

collect 操作符使用

collect 是Flow最基础的操作符，大多数场景下都会使用它来收集Flow中生产的数据信息

runBlocking {listOf(1,2,3,4,5).asFlow().collect { value -> println(value) }}

reduce 操作符使用

作用1： 用于数据求和

 private fun testReduce() {runBlocking {val num = listOf(1,2,3,4,5).asFlow().reduce { a, b -> a + b }println("求和的结果：$num")}}

作用2：用于求最大值

private fun testReduce1() {runBlocking {val maxNum = listOf(1, 5, 2, 4, 5, 6, 3).asFlow().reduce { a, b ->if (a > b) a else b}println("求集合中的最大值：${maxNum}")}}

fold 操作符使用

fold 也是用于计算的操作符，只不过fold可以设置累加器的初始值，与reduce相比 多一个设置初始值的操作

private fun testFold() {runBlocking {val sum = (1..5).asFlow().fold(100) { a, b -> a + b }println("设置初始值后的累加结果：$sum")}}

launchIn 操作符使用

launchIn用于指定协程作用域，如果单独看如下这块代码可能觉得launchIn好像没有什么作用，但是一旦多个流想要并发执行时，launchIn就可以起到它该有的作用，为流指定了协程作用域之后，相当于每个流开启了一个协程，每个流都在自己协程中运行，所以可以使用launchIn来实现并发的功能。

 val mDispatcher = Executors.newSingleThreadExecutor().asCoroutineDispatcher()private fun testLaunchIn() {val scope = CoroutineScope(mDispatcher)(1..5).asFlow().onEach { println(it) }.onCompletion { mDispatcher.close() }.launchIn(scope)}