协程中的流

Posted 2023-02-04 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