协程中的流

Posted 2021-09-14 ImportSUC

协程中的异步流 基础

一.什么是异步流

​ 可以连续返回多个值

​ 与集合的区别：集合可以返回多个计算好的值，但是只能一次性返回

​ 与Rxjava 的流是同一个概念

二.如何创建异步流

1.最基础的流构建器 flow{}

• flow { … } 构建块中的代码可以挂起,不再需要 suspend 修饰符。
• emit 发射， collect 收集。flow 构建器中的代码直到流被收集的时候才运行
• 不阻塞线程
• collect在哪个协程中调用，如果没有特殊操作，则flow代码块也在相同协程中运行

``

fun simple(): Flow<Int> = flow {
    println("Flow started")
    for (i in 1..3) {
        kotlinx.coroutines.delay(1000)
        emit(i)
    }
}

fun main() = runBlocking {
     println("Calling simple function...")
     var flow = simple()
     println("Calling collect...")
     flow.collect { value -> println(value) }
     println("Calling collect again...")
     flow.collect { value -> println(value) }
}

结果显示

Calling simple function...
Calling collect...
Flow started
1
2
3
Calling collect again...
Flow started
1
2
3
end

2.flowof 构建器

定义一个发射固定值集的流

private fun simple2() = runBlocking {
    flowOf(1, 2, 3).collect { value -> println(value) }
}

3. asFlow() 扩展函数

使用asFlow() 扩展函数，可以将各种结合转换为流

private fun simaple3() = runBlocking {
    listOf(1, 2, 3).asFlow().collect { value -> println(value) }
}

三.操作符

1.filter , map 等过渡操作符

应用于上游，作用于下游

suspend fun performRequest(request: Int): String {
    delay(1000) 
    return "response $request"
}

private suspend fun simple5() {
    listOf(1, 2, 3).asFlow().filter { it > 2 }.map { performRequest(it) }
    .collect { println(it) }
}

结果显示

response 3

2.take限长操作符

当满足条件时，Flow 会被取消，此时会抛出异常。需要用try{} 来处理

private fun simple4(): Flow<Int> = flow {
    try {
        emit(1)
        emit(2)
        println("This line will not execute")
        emit(3)
    } catch (e: Exception) {
        e.printStackTrace()
    } finally{
      
    }

}

simple4().take(1).collect { it -> println(it) }

结果显示：

1
kotlinx.coroutines.flow.internal.AbortFlowException: Flow was aborted, no more elements needed

3.末端流操作符

• 在流上用于启动流收集的挂起函数
• collect 是最基础的末端操作符
• toList 与 toSet。转化成各种集合
• first 取第一个值
• single 确保只发送一个值，否则就报错：java.lang.IllegalStateException: Expected only one element

fun simple6() = runBlocking {
    var first = listOf(1, 2, 3).asFlow().map { it + 1 }.first { it > 2 }
    println(first)
  
    var single = listOf(1).asFlow().map { it + 1 }.single()
    println(single)
}

4.flowOn 切换上下文

默认的，flow { ... } 构建器中的代码运行在相应流的收集器提供的上下文中

如果想让流运行在其他协程中，调用flowOn，创建一个新的协程供函数运行

fun simple7(): Flow<Int> = flow {
    for (i in 1..3) {
        Thread.sleep(100) 
        println("Emitting $i" + ",currentThread" + Thread.currentThread().name)
        emit(i)
    }
}.flowOn(Dispatchers.Default)

simple7().collect { value -> println("collect:" + value.toString().plus(",currentThread=" + Thread.currentThread().name)) }

结果显示

Emitting 1,currentThreadDefaultDispatcher-worker-1
collect:1,currentThread=main
Emitting 2,currentThreadDefaultDispatcher-worker-1
collect:2,currentThread=main
Emitting 3,currentThreadDefaultDispatcher-worker-1
collect:3,currentThread=main