kotlin flow操作符详解

implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core:1.3.3"
implementation "org.jetbrains.kotlinx:kotlinx-coroutines-android:1.3.3"
复制代码

suspend fun flow1() {
    flow<Int> {
        (0..4).forEach {
            emit(it)//生产者发送数据
        }
    }.collect {
        println(it)//消费者处理数据
    }
}
复制代码

flowOf(1, 2, 5, 4).collect {
        println(it)
    }
复制代码

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

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

flow<Int> {
        var userId = login()
        userId
    }.map {
        var permission = getPermission(it)
        permission
    }.collect{
        println("打印权限 $it")
    }
复制代码

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

(1..9).asFlow().take(3).collect {
        println(it)
    }
复制代码

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

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

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

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
复制代码

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

1
3
6
9
复制代码

1
2
3
4
5
6
7
8
9
复制代码

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

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

1 : 李白
2 : 杜甫
3 : 安安安安卓
复制代码

val flow1 = (1..5).asFlow()
复制代码

1 : 李白
2 : 杜甫
3 : 安安安安卓
复制代码

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 : 安安安安卓
复制代码

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 : 安安安安卓
复制代码

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)
复制代码

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
复制代码

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

1
23
5
3
4
刘操作完成
12344