网站建设打造,怎么给网站做百度坐标定位,怎样创建网站发招聘信息,部署wordpressFlow一、Flow1、Flow是什么东西#xff1f;2、实现功能3、特点4、冷流和热流5、流的连续性6、流的构建器7、流的上下文8、指定流所在协程9、流的取消9.1、超时取消9.2、主动取消9.3、密集型任务的取消10、背压和优化10.1、buffer 操作符10.2、 flowOn10.3、conflate 操作符10.…
Flow一、Flow1、Flow是什么东西2、实现功能3、特点4、冷流和热流5、流的连续性6、流的构建器7、流的上下文8、指定流所在协程9、流的取消9.1、超时取消9.2、主动取消9.3、密集型任务的取消10、背压和优化10.1、buffer 操作符10.2、 flowOn10.3、conflate 操作符10.4、collectLatest 操作符二、操作符1、变换操作符1.1、buffer 缓存1.2、map 变换1.2.1、map1.2.2、mapNotNull 不空的下发1.2.3、mapLatest1.3、transform 一转多1.4、reduce 累*加减乘除1.5、fold累*加减乘除 and 拼接1.6、flatMapConcat 有序变换1.7、flatMapMerge 无序变换1.8、flatMapLatest 截留2、过滤型操作符2.1、take 截留2.1.2、takeWhile2.2、filter满足条件下发2.2.2、filterNotNull 不空的下发2.2.3、filterNot符合条件的值将被丢弃2.2.4、filterInstance 筛选符合类型的值2.3、skip 和 drop跳过2.3.2、dropWhile2.4、distinctUntilChanged 过滤重复2.4.2、distinctUntilChangedBy2.5、single 判断是否一个事件2.6、first 截留第一个事件2.7、debounce 防抖动2.8、conflate2.9、sample 周期采样3、组合型操作符3.1、count 计数3.2、zip 合并元素3.3、combine合并元素3.4、merge 合并成流3.5、flattenConcat 展平流3.6、flattenMerge展平流4、异常操作符4.1、catch 拦截异常4.2、retry 重试4.2.2、retryWhen4.3、withTimeout 超时5、辅助操作符5.1、onXXX5.2、delay 延时5.3、measureTimeMillis 计时参考地址一、Flow
1、Flow是什么东西
Flow 是有点类似 RxJava 的 Observable
都有冷流和热流之分 都有流式构建结构 都包含 map、filter 等操作符。
区别于ObservableFlow可以配合挂起函数使用
2、实现功能
异步返回多个值
可以实现下载功能等Observable 下发数组时可以实现什么功能他就能实现什么功能 当文件下载时对应的后台下载进度就可以通过Flow里面的emit发送数据通过collect接收对应的数据。
转https://blog.csdn.net/qq_30382601/article/details/121825461
3、特点
flow{…}块中的代码可以挂起使用flowsuspend修饰符可以省略流使用emit函数发射值流使用collect的函数收集值flow类似冷流flow中代码直到流被收集(调用collect)的时候才运行类似lazy什么时候用什么时候执行。流的连续性流收集都是按顺序收集的flowOn可更改流发射的上下文即可以指定在主线程或子线程中执行与之相对的是热流我们即将介绍的 StateFlow 和 SharedFlow 是热流在垃圾回收之前都是存在内存之中并且处于活跃状态的。
转https://blog.csdn.net/zx_android/article/details/122744370
4、冷流和热流 冷流 冷流类似冷启动代码在被用到才会执行如你需要使用的数据在网络需要先请求网络才能得到数据 Flow是一种类似于序列的冷流flow构建器中的代码直到流被收集的时候才运行。 热流 热流类似热启动代码在用到之前已经准备好如你请求过网络数据已经缓存在本地你只需直接使用即可
5、流的连续性
流的连续性流收集都是按顺序收集的 6、流的构建器
如下三种为冷流构建器
flow{emit} .collect{}flowOf***.collect{}***.asFlow().collect{} Testfun test flow builder() runBlockingUnit {flowOf(one, two, three).onEach { delay(1000) }.collect { value -println(value)}(1..3).asFlow().collect { value -println(value)}flowInt {for (i in 11..13) {delay(1000) //假装在一些重要的事情emit(i) //发射产生一个元素}}.collect { value -println(value)}}7、流的上下文
flowOn 多用于切线程
流的收集总是在调用协程的上下文中发生流的该属性称为上下文保存。 fun simpleFlow3() flowInt {println(Flow started ${Thread.currentThread().name})for (i in 1..3) {delay(1000)emit(i)}}Testfun test flow context() runBlockingUnit {simpleFlow3().collect { value - println(Collected $value ${Thread.currentThread().name}) }}如下流的发射和接收在一个协程内
Flow started Test worker coroutine#1
Collected 1 Test worker coroutine#1
Collected 2 Test worker coroutine#1
Collected 3 Test worker coroutine#1flow{…}构建器中的代码必须遵循上下文保存属性并且不允许从其他上下文中发生(emit)
如下这种写法不被允许 fun simpleFlow4() flowInt {withContext(Dispatchers.Default) {println(Flow started ${Thread.currentThread().name})for (i in 1..3) {delay(1000)emit(i)}}}那么如何切换协程上下文呢 flowOn操作符该函数用于更改流发射的上下文 fun simpleFlow5() flowInt {println(Flow started ${Thread.currentThread().name})for (i in 1..3) {delay(1000)emit(i)}}.flowOn(Dispatchers.Default)Testfun test flow context() runBlockingUnit {simpleFlow5().collect { value - println(Collected $value ${Thread.currentThread().name}) }}如下切换上下文成功
Flow started DefaultDispatcher-worker-2 coroutine#2
Collected 1 Test worker coroutine#1
Collected 2 Test worker coroutine#1
Collected 3 Test worker coroutine#18、指定流所在协程
launchIn 用于指定协程作用域通知flow执行
使用 launchIn 替换 collect 在单独的协程中启动收集流
指定协程 //事件源private fun events() (1..3).asFlow().onEach { delay(100) }.flowOn(Dispatchers.Default)Testfun test flow launch() runBlockingUnit {val job events().onEach { event - println(Event: $event ${Thread.currentThread().name}) }
// .collect {}.launchIn(CoroutineScope(Dispatchers.IO)) //这里使用另一个上下文传入Flow
// .launchIn(this)//这里使用当前上下文传入Flowjob.join()}打印
Event: 1 DefaultDispatcher-worker-2 coroutine#2
Event: 2 DefaultDispatcher-worker-1 coroutine#2
Event: 3 DefaultDispatcher-worker-3 coroutine#2也可以指定当前协程中执行 Testfun test flow launch() runBlockingUnit {val job events().onEach { event - println(Event: $event ${Thread.currentThread().name}) }
// .collect {}
// .launchIn(CoroutineScope(Dispatchers.IO)) //这里使用另一个上下文传入Flow.launchIn(this)//这里使用当前上下文传入Flow// job.join()}Event: 1 Test worker coroutine#2
Event: 2 Test worker coroutine#2
Event: 3 Test worker coroutine#29、流的取消
流采用和协程同样的协作取消。流可以在挂起函数的挂起的时候取消。
9.1、超时取消
withTimeoutOrNull 不能取消密集型任务 fun simpleFlow6() flowInt {for (i in 1..300) {delay(1000)emit(i)println(Emitting $i)}}Testfun test cancel flow() runBlockingUnit {withTimeoutOrNull(2500) {simpleFlow6().collect { value - println(value) }}println(Done)}9.2、主动取消
cancel Testfun test cancel flow () runBlockingUnit {simpleFlow6().collect { value -if (value 3) {cancel()}println(value)}println(Done)}9.3、密集型任务的取消
密集型任务需要流的取消检测
cancel cancellable Testfun test cancel flow check() runBlockingUnit {(1..5).asFlow().cancellable().collect { value -println(value)if (value 3) cancel()println(cancel check ${coroutineContext[Job]?.isActive})}}10、背压和优化
什么是背压 生产者生产的效率大于消费者消费的效率元素积压
例演示背压 fun simpleFlow8() flowInt {for (i in 1..10) {// emit 上面这段代码在collect之前执行delay(100)emit(i) // 调用collect// emit下面这段代码在 collect 之后执行println(Emitting $i ${Thread.currentThread().name})}}Testfun test flow back pressure() runBlockingUnit {val time measureTimeMillis {simpleFlow8().collect { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}Collected 1 Test worker coroutine#1
Emitting 1 Test worker coroutine#1
Collected 2 Test worker coroutine#1
Emitting 2 Test worker coroutine#1
Collected 3 Test worker coroutine#1
Emitting 3 Test worker coroutine#1
Collected 4 Test worker coroutine#1
Emitting 4 Test worker coroutine#1
Collected 5 Test worker coroutine#1
Emitting 5 Test worker coroutine#1
Collected 6 Test worker coroutine#1
Emitting 6 Test worker coroutine#1
Collected 7 Test worker coroutine#1
Emitting 7 Test worker coroutine#1
Collected 8 Test worker coroutine#1
Emitting 8 Test worker coroutine#1
Collected 9 Test worker coroutine#1
Emitting 9 Test worker coroutine#1
Collected 10 Test worker coroutine#1
Emitting 10 Test worker coroutine#1
Collected in 3169 ms如何解决背压
通过缓存进行性能优化
10.1、buffer 操作符
并发运行流中发射元素的代码
注意for (i in 1…10) 这里用的是 1到 10原因是 for循环 有耗时问题通过打印时间戳在 for (i in 1…x) 上下发现 for (i in 1…x) 这行代码有时耗时超过200毫秒目前不知是何问题特此记录为方便对比优化时长使用1到10. Testfun test flow back pressure buffer() runBlockingUnit {val time measureTimeMillis {simpleFlow8().buffer(10) //缓存发射事件.collect { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}Emitting 1 Test worker coroutine#2
Emitting 2 Test worker coroutine#2
Collected 1 Test worker coroutine#1
Emitting 3 Test worker coroutine#2
Emitting 4 Test worker coroutine#2
Collected 2 Test worker coroutine#1
Emitting 5 Test worker coroutine#2
Emitting 6 Test worker coroutine#2
Collected 3 Test worker coroutine#1
Emitting 7 Test worker coroutine#2
Emitting 8 Test worker coroutine#2
Collected 4 Test worker coroutine#1
Emitting 9 Test worker coroutine#2
Emitting 10 Test worker coroutine#2
Collected 5 Test worker coroutine#1
Collected 6 Test worker coroutine#1
Collected 7 Test worker coroutine#1
Collected 8 Test worker coroutine#1
Collected 9 Test worker coroutine#1
Collected 10 Test worker coroutine#1
Collected in 2398 ms10.2、 flowOn
flowOn(修改流上下文达到异步处理的效果从而优化背压 Testfun test flow back pressure flowOn() runBlockingUnit {val time measureTimeMillis {simpleFlow8().flowOn(Dispatchers.IO).collect { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}Emitting 1 DefaultDispatcher-worker-1 coroutine#2
Emitting 2 DefaultDispatcher-worker-1 coroutine#2
Collected 1 Test worker coroutine#1
Emitting 3 DefaultDispatcher-worker-1 coroutine#2
Emitting 4 DefaultDispatcher-worker-1 coroutine#2
Collected 2 Test worker coroutine#1
Emitting 5 DefaultDispatcher-worker-1 coroutine#2
Emitting 6 DefaultDispatcher-worker-1 coroutine#2
Collected 3 Test worker coroutine#1
Emitting 7 DefaultDispatcher-worker-1 coroutine#2
Emitting 8 DefaultDispatcher-worker-1 coroutine#2
Collected 4 Test worker coroutine#1
Emitting 9 DefaultDispatcher-worker-1 coroutine#2
Emitting 10 DefaultDispatcher-worker-1 coroutine#2
Collected 5 Test worker coroutine#1
Collected 6 Test worker coroutine#1
Collected 7 Test worker coroutine#1
Collected 8 Test worker coroutine#1
Collected 9 Test worker coroutine#1
Collected 10 Test worker coroutine#1
Collected in 2385 ms10.3、conflate 操作符
conflate()合并发射项处理最新的值不对每个值进行处理 Testfun test flow back pressure conflate() runBlockingUnit {val time measureTimeMillis {simpleFlow8().conflate().collect { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}Emitting 1 Test worker coroutine#2
Emitting 2 Test worker coroutine#2
Collected 1 Test worker coroutine#1
Emitting 3 Test worker coroutine#2
Emitting 4 Test worker coroutine#2
Collected 2 Test worker coroutine#1
Emitting 5 Test worker coroutine#2
Emitting 6 Test worker coroutine#2
Collected 4 Test worker coroutine#1
Emitting 7 Test worker coroutine#2
Emitting 8 Test worker coroutine#2
Collected 6 Test worker coroutine#1
Emitting 9 Test worker coroutine#2
Emitting 10 Test worker coroutine#2
Collected 8 Test worker coroutine#1
Collected 10 Test worker coroutine#1
Collected in 1554 ms10.4、collectLatest 操作符
collectLatest()取消并重新发射最后一个值 Testfun test flow back pressure collectLatest() runBlockingUnit {val time measureTimeMillis {simpleFlow8().collectLatest { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}Emitting 1 Test worker coroutine#2
Emitting 2 Test worker coroutine#2
Emitting 3 Test worker coroutine#2
Emitting 4 Test worker coroutine#2
Emitting 5 Test worker coroutine#2
Emitting 6 Test worker coroutine#2
Emitting 7 Test worker coroutine#2
Emitting 8 Test worker coroutine#2
Emitting 9 Test worker coroutine#2
Emitting 10 Test worker coroutine#2
Collected 10 Test worker coroutine#12
Collected in 1648 ms二、操作符
1、变换操作符
1.1、buffer 缓存
上面背压有栗子
1.2、map 变换
1.2.1、map
map 是变换元素
data class Student(var name: String, var age: Int)private suspend fun performRequest(age: Int): Student {delay(500)return Student(这是name, age)}Testfun test map flow operator() runBlockingUnit {(1..3).asFlow().map { request - performRequest(request) }.collect { value - println(value) }}Student(name这是name, age1)
Student(name这是name, age2)
Student(name这是name, age3)1.2.2、mapNotNull 不空的下发 Testfun test mapNotNull flow operator() runBlockingUnit {flow {emit(1)emit(3)emit(2)}.mapNotNull { request -if (1 request) {null} else {Student(这是name, request)}}.collect { value - println(value) }}Student(name这是name, age3)
Student(name这是name, age2)1.2.3、mapLatest
当有新值发送时如果上个转换还没结束会取消掉用法同map Testfun test mapLatest flow operator() runBlockingUnit {flow {emit(1)emit(2)emit(3)}.mapLatest {if (2 it) delay(100L)it is $it}.collect {println(it)}}1.3、transform 一转多 Testfun test transform flow operator() runBlockingUnit {(1..3).asFlow().transform { request -emit(Making request $request)emit(performRequest(request))}.collect { value - println(value) }}Making request 1
Student(name这是name, age1)
Making request 2
Student(name这是name, age2)
Making request 3
Student(name这是name, age3)1.4、reduce 累*加减乘除 Testfun test reduce operator() runBlockingUnit {println(flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.reduce { accumulator, value - accumulator value })}1.5、fold累*加减乘除 and 拼接
加 Testfun test fold operator() runBlockingUnit {println(flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.fold(3) { accumulator, value - accumulator value })}17减 Testfun test fold - operator() runBlockingUnit {println(flowInt {emit(2)emit(3)}.fold(18) { accumulator, value - accumulator - value })}13乘 Testfun test fold multiply by operator() runBlockingUnit {println(flowInt {emit(1)emit(1)emit(2)emit(3)}.fold(3) { accumulator, value - accumulator * value })}18除 Testfun test fold devide operator() runBlockingUnit {println(flowInt {emit(2)emit(3)}.fold(18) { accumulator, value - accumulator / value })}3拼接 Testfun test fold joint operator() runBlockingUnit {println(flowInt {emit(1)emit(2)emit(3)}.fold(拼接) { accumulator, value - returnfold $accumulator $value })}拼接 1 2 31.6、flatMapConcat 有序变换
元素会变换完以流的形式继续下发并且某个元素需要耗时它后面的元素会等待。 Testfun test flatMapConcat operator() runBlockingUnit {(1..5).asFlow().onEach { delay(100) }.flatMapConcat { num -flow {if (3num){delay(200)}emit(num: $num)}}.collect {println(value - $it)}}value - num: 1
value - num: 2
value - num: 3
value - num: 4
value - num: 51.7、flatMapMerge 无序变换
元素会变换完以流的形式继续下发并且某个元素需要耗时它后面的元素不会等待。 Testfun test flatMapMerge operator() runBlockingUnit {(1..5).asFlow().onEach { delay(100) }.flatMapMerge() { num -flow {if (3num){delay(200)}emit(num: $num)}}.collect {println(value - $it)}}value - num: 1
value - num: 2
value - num: 4
value - num: 3
value - num: 51.8、flatMapLatest 截留
快速执行的事件都正常下发 当有新值发送时如果上个转换还没结束会上取消掉上一个直接下发新值。 Testfun test flatMapLatest operator() runBlockingUnit {(1..5).asFlow().onEach { delay(100) }.flatMapLatest() { num -flow {if (3 num) {delay(200)}emit(num: $num)emit(num2: $num)}}.collect {println(value - $it)}}value - num: 1
value - num2: 1
value - num: 2
value - num2: 2
value - num: 4
value - num2: 4
value - num: 5
value - num2: 52、过滤型操作符
2.1、take 截留
跟Rxjava一样 fun numbers() flowInt {try {emit(1)emit(2)println(This line will not execute)emit(3)} finally {println(Finally in numbers)}}Testfun test limit length operator() runBlockingUnit {//take(2),表示 当计数元素被消耗时原始流被取消numbers().take(2).collect { value - println(value) }}1
2
Finally in numbers2.1.2、takeWhile
找到第一个不满足条件的值发送它之前的值和dropWhile相反 Testfun test takeWhile operator() runBlockingUnit {flowInt {emit(2)emit(1)emit(3)emit(4)emit(1)}.takeWhile { it 2 }.collect { value - println(value) }}如上什么也不会输出 Testfun test takeWhile operator() runBlockingUnit {flowInt {emit(1)emit(2)emit(3)emit(4)emit(1)}.takeWhile { it 2 }.collect { value - println(value) }}
会输出 1
2.2、filter满足条件下发
跟Rxjava一样 Testfun test filter operator() runBlockingUnit {numbers().filter {it 2}.collect { value - println(value) }}2.2.2、filterNotNull 不空的下发 Testfun test filterNotNull flow operator() runBlockingUnit {flow {emit(1)emit(3)emit(null)emit(2)}.filterNotNull ().collect { value - println(value) }}1
3
22.2.3、filterNot符合条件的值将被丢弃
筛选不符合条件的值相当于filter取反 Testfun test filterNot operator() runBlockingUnit {flowInt {emit(1)emit(2)emit(3)}.filterNot {it 2}.collect { value - println(value) }}1
22.2.4、filterInstance 筛选符合类型的值
对标rxjava中的ofType
筛选符合类型的值不符合类型的值将被丢弃 Testfun test filterInstance operator() runBlockingUnit {flowAny {emit(1)emit(2)emit(3)emit(str)}.filterIsInstanceString().collect { value - println(value) }}2
str2.3、skip 和 drop跳过 Testfun test skip operator() runBlockingUnit {numbers().drop(2).collect { value - println(value) }}输出
32.3.2、dropWhile
找到第一个不满足条件的值继续发送它和它之后的值 Testfun test dropWhile operator() runBlockingUnit {numbers().dropWhile { it 2 }.collect { value - println(value) }}This line will not execute
3
Finally in numbers2.4、distinctUntilChanged 过滤重复 Testfun test distinctUntilChanged operator() runBlockingUnit {flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.distinctUntilChanged().collect { value - println(value) }}2.4.2、distinctUntilChangedBy
判断两个连续值是否重复可以设置是否丢弃重复值。 去重规则有点复杂没完全懂 Testfun test distinctUntilChangedBy operator() runBlockingUnit {flowOf(Student(name Jack, age 11),Student(name Tom, age 10),Student(name Jack, age 12),Student(name Jack, age 13),Student(name Tom, age 11)).distinctUntilChangedBy { it.name Jack }.collect { //第三个Stu将被丢弃println(it.toString())}}Student(nameJack, age11)
Student(nameTom, age10)
Student(nameJack, age12)
Student(nameTom, age11)2.5、single 判断是否一个事件
用于确保 flow 输出值唯一。若只有一个值则可以正常执行若输出的值不止只有一个的时候就会抛出异常 Testfun test single operator() runBlockingUnit {try {println(flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.single())} catch (e: Exception) {println(e $e)}}如果一个事件就正常执行否则异常。
e java.lang.IllegalArgumentException: Flow has more than one element2.6、first 截留第一个事件 Testfun test first operator() runBlockingUnit {println(flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.first())}12.7、debounce 防抖动 Testfun test debounce operator() runBlockingUnit {flowOf(Student(name Jack, age 11),Student(name Tom, age 10),Student(name Jack, age 12),Student(name Jack, age 13),Student(name Tom, age 11)).onEach {if (it.name Jack it.age 13)delay(500)}.debounce(500).collect { //第三个Stu将被丢弃println(it.toString())}}Student(nameJack, age12)
Student(nameTom, age11)2.8、conflate
见 10.3、conflate
仅保留最新值, 内部就是 buffer(CONFLATED)
2.9、sample 周期采样
固定周期采样 给定一个时间周期保留周期内最后发出的值其他的值将被丢弃
sample操作符与debounce操作符有点像但是却限制了一个周期性时间sample操作符获取的是一个周期内的最新的数据可以理解为debounce操作符增加了周期的限制。 Testfun test sample operator() runBlockingUnit {flow {repeat(10) {delay(50)emit(it)}}.sample(100).collect {println(it)}}0
2
4
6
83、组合型操作符
3.1、count 计数 Testfun test count operator() runBlockingUnit {println(flowInt {emit(1)emit(1)emit(2)emit(3)emit(3)emit(4)}.count())}3.2、zip 合并元素
跟Rxjava一样 Testfun test zip operator() runBlockingUnit {val nameFlow mutableListOf(小红, 小黑).asFlow()val numFlow (1..3).asFlow()nameFlow.zip(numFlow) { string, num -$string$num}.collect {println(value - $it)}}3.3、combine合并元素 Testfun test combine operator() runBlockingUnit {val nameFlow mutableListOf(小红, 小黑).asFlow()val numFlow (1..3).asFlow()nameFlow.combine(numFlow) { string, num -$string$num}.collect {println(value - $it)}}value - 小红1
value - 小黑2
value - 小黑33.4、merge 合并成流
merge 是将两个flow合并起来将每个值依次发出来 Testfun test merge operator() runBlockingUnit {val flow1 listOf(1, 2).asFlow()val flow2 listOf(one, two, three).asFlow()merge(flow1, flow2).collect { value - println(value) }}1
2
one
two
three3.5、flattenConcat 展平流
展平操作符 flattenConcat 以顺序方式将给定的流展开为单个流通俗点讲减少层级 感觉和merge这么像呢这个不太理解啥用 Testfun test flattenConcat operator() runBlockingUnit {val flow1 listOf(1, 2).asFlow()val flow2 listOf(one, two, three).asFlow()val flow3 listOf(x, xx, xxx).asFlow()flowOf(flow1, flow2, flow3).flattenConcat().collect { value - println(value) }}1
2
one
two
three
x
xx
xxx3.6、flattenMerge展平流
flattenMerge 作用和 flattenConcat 一样但是可以设置并发收集流的数量 Testfun test flattenMerge operator() runBlockingUnit {val flow1 listOf(1, 2).asFlow()val flow2 listOf(one, two, three).asFlow()val flow3 listOf(x, xx, xxx).asFlow()flowOf(flow1, flow2, flow3).flattenMerge(2).collect { value - println(value) }}1
2
one
two
three
x
xx
xxx4、异常操作符
4.1、catch 拦截异常
对标rxjava 中的 onErrorResumeNext
Exception、Throwable、Error 都会拦截 Testfun test catch operator() runBlockingUnit {(1..5).asFlow().onEach { delay(100) }.onEach { if (2 it) throw NullPointerException() }.catch {emit(110)println(e $it)}.collect {println(value - $it)}}Testfun test catch operator() runBlockingUnit {(1..5).asFlow().onEach { delay(100) }.onEach { if (2 it)
// throw Exception(测试 异常)
// throw Throwable(测试 异常)throw Error(测试 错误)}.catch {emit(110)println(e $it)}.collect {println(value - $it)}}value - 1
value - 110
e java.lang.Error: 测试 错误4.2、retry 重试
所有异常错误都拦截
拦截次数 Testfun test retry operator() runBlockingUnit {flowAny {emit(1)emit(2)throw Exception(异常)emit(3)}.retry(2).catch { emit(110) }.collect { value - println(value) }}拦截条件 Testfun test retry 2 operator() runBlockingUnit {flowAny {emit(1)emit(2)throw Error(异常)emit(3)}.retry { it.message 异常 }.catch { emit(110) }.collect { value - println(value) }}如上满足拦截条件所以会一直打印日志
1
2
1
2
1
2
1
2
1
... 不杀死程序一直打印4.2.2、retryWhen
4.3、withTimeout 超时 Testfun test retry 2 operator() runBlockingUnit {withTimeout(2500) {flowAny {emit(1)throw Error(异常)}.retry { it.message 异常 }.catch { emit(110) }.collect { value - println(value) }}}输出
1
1
... 好多个
1
1
1Timed out waiting for 2500 ms
kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 2500 ms(Coroutine boundary)at com.yoshin.kt.kotlindemo20220713.ExampleUnitTest$test retry 2 operator$1.invokeSuspend(ExampleUnitTest.kt:928)
Caused by: kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 2500 msat app//kotlinx.coroutines.TimeoutKt.TimeoutCancellationException(Timeout.kt:184)at app//kotlinx.coroutines.TimeoutCoroutine.run(Timeout.kt:154)at app//kotlinx.coroutines.EventLoopImplBase$DelayedRunnableTask.run(EventLoop.common.kt:508)at app//kotlinx.coroutines.EventLoopImplBase.processNextEvent(EventLoop.common.kt:284)at app//kotlinx.coroutines.DefaultExecutor.run(DefaultExecutor.kt:108)at java.base11.0.13/java.lang.Thread.run(Thread.java:834)5、辅助操作符
5.1、onXXX
onXXX 的方法包含
onCompletion 流完成时调用 onStart 流开始时调用 onEach 元素下发时调用每次下发都调用
对比rxjava 中 onCompletion doOnComplete onStart doOnSubscribe 或者 doOnLifecycle onEach doNext Testfun test do operator() runBlockingUnit {(1..5).asFlow().onCompletion { println( onCompletion $it ) }.onStart { println( onStart ) }.onEach { println( onEach $it ) }.collect {println(value - $it)}}5.2、delay 延时
延时 private fun events() (1..3).asFlow().onEach { delay(100) }.flowOn(Dispatchers.Default)5.3、measureTimeMillis 计时
测量代码用时 Testfun test flow back pressure() runBlockingUnit {val time measureTimeMillis {simpleFlow8().collect { value -delay(200) //处理这个元素消耗 200msprintln(Collected $value ${Thread.currentThread().name})}}println(Collected in $time ms)}参考地址
笔记大部分内容来自动脑学院的文章和视频
动脑学院 https://blog.csdn.net/qq_30382601/article/details/121825461
Kotlin 之 协程三Flow异步流 https://blog.csdn.net/zx_android/article/details/122744370
Android Kotlin之Flow数据流https://blog.csdn.net/u013700502/article/details/120526170
