一、Flow通过flow异步返回多个值fun simpleFlow() flowInt { for (i in 1..5) { delay(1000) emit(i) } } Test fun 通过flow异步返回多个值() runBlocking { launch { for (k in 1..5) { delay(500) println(k:$k) } } delay(500) simpleFlow().collect { println(it) } println(end) delay(5000) }执行结果k:1 k:2 1 k:3 k:4 2 k:5 3 4 5 end冷流Flow是一种类似于序列的冷流flow构建器中的代码直到遇到末端操作符(例如被收集)的时候才会运行。Test fun 冷流在收集时才会开始执行()runBlocking { val flowsimpleFlow() println(创建冷流) delay(1000) println(冷流未执行) flow.collect { println(it) } println(冷流执行结束) }执行结果创建冷流 冷流未执行 1 2 3 4 5 冷流执行结束流构建器flowOf构建器定义了一个发射固定值集的流。Test fun 流构建器()runBlocking { flowOf(1,2,3) .onEach { delay(1000) } .collect { println(it) } }流上下文流的收集总是在调用协程的上下文中发生Flow 发射、操作符执行、收集时能正确继承 / 保留你指定的协程上下文。流的该属性称为上下文保存。fun contextFlow()flow{ println(Thread.currentThread().name) for (i in 1..5) { delay(1000) emit(i) } } Test fun 流的上下文保存()runBlocking { contextFlow().collect { println($it : ${Thread.currentThread().name}) } }执行结果flow运行的线程与协程的相同。Test worker coroutine#1 1 : Test worker coroutine#1 2 : Test worker coroutine#1 3 : Test worker coroutine#1 4 : Test worker coroutine#1 5 : Test worker coroutine#1flow{...}构建器中的代码必须遵循上下文保存属性并且不允许从其他上下文中发射(emit)。例如不可以使用withContext来切换上下文。flow { withContext(Dispatchers.IO) { // 报错 emit(1) } }flowOn操作符该函数用于更改流发射的上下文。fun contextFlow()flow{ println(Thread.currentThread().name) for (i in 1..5) { //模拟IO操作 delay(1000) emit(i) } }.flowOn(Dispatchers.IO) //上下文切换 Test fun 流的上下文保存()runBlocking { contextFlow().collect { println($it : ${Thread.currentThread().name}) } }执行结果flow运行的线程与协程的不同。DefaultDispatcher-worker-1 coroutine#2 1 : Test worker coroutine#1 2 : Test worker coroutine#1 3 : Test worker coroutine#1 4 : Test worker coroutine#1 5 : Test worker coroutine#1flowOn只作用于他上面的代码下面的代码还是运行在上下文中的线程。Test fun flowOn只作用其上面的代码()runBlocking { flowOf(1,2,3) .onEach { delay(500) println(flow上面$it ${Thread.currentThread().name}) } .flowOn(Dispatchers.IO) .onEach { delay(500) println(flow下面$it ${Thread.currentThread().name}) } .collect { println(收集元素$it ${Thread.currentThread().name}) } }执行结果flow上面的代码运行在IO线程flow下面的代码运行在协程中。若想要更直观的结果可以将第二个onEach中的delay(500)删掉这样可以使接收速度大于发送速度这样子的话结果将会是flow上面、flow下面、收集元素这样的顺序出现。flow上面1 DefaultDispatcher-worker-1 coroutine#2 flow上面2 DefaultDispatcher-worker-1 coroutine#2 flow下面1 Test worker coroutine#1 收集元素1 Test worker coroutine#1 flow上面3 DefaultDispatcher-worker-1 coroutine#2 flow下面2 Test worker coroutine#1 收集元素2 Test worker coroutine#1 flow下面3 Test worker coroutine#1 收集元素3 Test worker coroutine#1在指定协程中收集流fun event()(1..3) .asFlow() .onEach { delay(1000) } .flowOn(Dispatchers.Default) Test fun 在指定协程中收集流()runBlocking { val scope CoroutineScope(Dispatchers.IO) val jobevent().onEach { println(Event:$it ${Thread.currentThread().name}) } .launchIn(scope) job.join() }执行结果虽然不是在同一个线程中运行的但还是同一个协程。Event:1 DefaultDispatcher-worker-3 coroutine#2 Event:2 DefaultDispatcher-worker-1 coroutine#2 Event:3 DefaultDispatcher-worker-3 coroutine#2launchIn相当于一个末端操作符相当于在对应协程中用launch启动一个子协程子协程中有流的收集(没有末端操作符流不会执行)下面是它的源码。public fun T FlowT.launchIn(scope: CoroutineScope): Job scope.launch { collect() // tail-call }流的超时取消fun 流的超时取消--产生流() flowInt{ for (i in 1..3){ emit(i) println(发送数据$i) } } Test fun 流的超时取消()runBlocking { withTimeoutOrNull(2500){ 流的超时取消--产生流().collect { delay(1000) println(收集数据$it) } } println(等待超时) }执行结果emit()是挂起函数如果buffer为0(或者未设置buffer,下面处理背压中有示例展示设置buffer的情况)它会等待collect中的代码执行完成之后再恢复。所以会出现收集数据先于发送数据。收集数据1 发送数据1 收集数据2 发送数据2 等待超时流的取消检测Test fun 流没有正常取消()runBlocking { (1..5).asFlow().collect { println(it) if(it3) cancel() } }执行结果出现取消异常但是取消异常是在5之后说明没有正常地取消流。1 2 3 4 5添加cancellable()Test fun 流没有正常取消()runBlocking { (1..5).asFlow().cancellable().collect { println(it) if(it3) cancel() } }执行到3之后就会出现取消异常说明正确地取消流。1 2 3用emit发射数据时emit函数会自动检测流是否被取消。使用缓冲与flowOn处理背压当生产者生产速度大于消费者消费速度时会产生背压。设置buffer解决背压fun 背压-发射数据()flowInt{ for(i in 1..3){ delay(500) emit(i) println(发射$i) } } Test fun 设置buffer解决背压()runBlocking { 背压-发射数据() .buffer(3) .collect { delay(1000) println(接收$it) } }执行结果发射1 发射2 接收1 发射3 接收2 接收3用flowOn解决背压Test fun 使用folwOn解决背压()runBlocking { 背压-发射数据() .flowOn(Dispatchers.Default) .collect { delay(1000) println(接收$it) } }执行结果发射1 发射2 接收1 发射3 接收2 接收3流的超时取消这一小节中有提到buffer。当flow上下文与协程上下文相同时buffer默认为0只有当数据收集之后才可以发射下一个数据当用flowOn切换上下文时会自动创建一个不为0的buffer从而解决背压问题。flow除了设置buffer还有其他的一些设置。conflate(),可收集最新的数据中间发射的数据不收集collectLatest(),只收集最后一个值。操作符转换操作符transformTest fun transform转换操作符()runBlocking { (1..3).asFlow() .transform { emit(it) emit(it5) }.collect { println(it) } }1 6 2 7 3 8map会自动返回最后的lambda表达式transform要自己手动emit。限长操作符takefun take-发射数据()flow { try { emit(1) emit(2) println(不会发射) emit(3) }finally { println(未全部发射) } } Test fun take限长操作符()runBlocking { take-发射数据().take(2).collect { println(it) } }1 2 未全部发射末端操作符末端操作符是在流上用于启动流收集的函数。collect、toList、toSet、first、single、reduce、fold。组合操作符zipTest fun zip组合操作符()runBlocking { val nums(1..3).asFlow().onEach { delay(300) } val strflowOf(one,two,three).onEach { delay(400) } val startTime System.currentTimeMillis() nums.zip(str){ nums,str - $nums $str }.collect { println(${System.currentTimeMillis()-startTime} : $it) } }435 : 1 one 835 : 2 two 1244 : 3 three展平操作符展平操作符就是专门处理流里套流FlowFlowT → 展开成普通流FlowTTest fun 展平操作符()runBlocking { val startTime System.currentTimeMillis() (1..3).asFlow() .onEach { delay(100) } .flatMapConcat { //其它展开操作符修改这一行就行 flow { emit(it) delay(1000) emit((it10)) } } .collect { println(${System.currentTimeMillis()-startTime} : 收集 ${it}) } }flatMapConcat126 : 收集 1 1147 : 收集 11 1255 : 收集 2 2273 : 收集 12 2375 : 收集 3 3396 : 收集 13flatMapMerge152 : 收集 1 251 : 收集 2 361 : 收集 3 1158 : 收集 11 1267 : 收集 12 1363 : 收集 13flatMapLatest140 : 收集 1 289 : 收集 2 396 : 收集 3 1407 : 收集 13流的异常捕获try/catch块Test fun try catch块捕获异常()runBlocking{ try { simpleFlow2().collect { println(it) check(it2) } }catch (e: Exception){ e.printStackTrace() } }catch函数Test fun catch函数捕获异常()runBlocking { flow{ emit(1) delay(200) emit(2) throw IOException() delay(200) emit(3) }.catch { println(it) }.collect { println(it) } }1 2流的完成当流收集完成时普通情况或异常情况它可以需要执行一个动作。命令式finally块