Android面试题之Kotlin异步流、冷流Flow

2024-06-13 20:52:10 浏览数 (1)

返回多个值

代码语言：javascript复制
//返回了多个值，是同步
fun simpleSequence():Sequence<Int> = sequence {
    for (i in 1..3){
        Thread.sleep(1000)//阻塞
        //因为有RestrictsSuspension注解，无法调用挂起函数，所以delay会报错
        //delay(1000)
        yield(i)
    }
}
通过Flow异步返回多个值
代码语言：javascript复制
fun simpleFlow() = flow<Int>{
    println("flow started")
    for (i in 1..3){
        delay(1000)
        emit(i)
    }
}

@Test
fun `test flow`() = runBlocking {
    val flow = simpleFlow()
    println("calling collect...")
    flow.collect {value -> println(value)}
    println("calling collect again...")
    flow.collect {value -> println(value)}
}
Flow与其他方式的区别
  • 名为flow的是Flow类型构建器函数
  • flow{...}构建块中的代码可以挂起
  • 函数simpleFlow不再有suspend修饰符
  • 流使用emit函数发射值，使用collect函数收集值
冷流
  • Flow是一种类似于序列的冷流，flow构建器中的代码直到流被收集的时候才运行（调用collect的时候）
  • 调用Collect以后，发射出来的值才会实实在在的存在于内存之中，和懒加载有点像
流的连续性
  • 流的每次单独收集都是按照顺序执行的，除非使用特殊的操作符
  • 从上游到下游每个过渡操作符都会处理每个发射出的值，然后再交给末端操作符
代码语言：javascript复制
//会输出string 2；string 4
@Test
fun `test flow2`() = runBlocking {
    (1..5).asFlow().filter { 
        it % 2 == 0
    }.map { 
        "string %it"
    }.collect { 
        println("collect $it")
    }
}
流构建器
  • flowOf构建器定义了一个发射固定值集的流
  • 使用.asFlow扩展函数可以将各种集合 与序列转换为流
流上下文
  • 流的收集总是在调用协程的上下文中发生，这个属性称为上下文保存
  • flow{...}构建器中的代码必须遵循上下文保存属性，并且不允许从其他上下文中发射（emit）
  • flowOn操作符，该函数用于更改流发射的上下文
代码语言：javascript复制
fun simpleFlow() = flow<Int>{
    println("flow started")
    for (i in 1..3){
        delay(1000)
        emit(i)
    }
}.flowOn(Dispatchers.IO)
启动流

使用launchIn替换collect，我们可以在单独的协程中启动流的收集

代码语言：javascript复制
fun events() = (1..3)
    .asFlow()
    .onEach { delay(100) }
    .flowOn(Dispatchers.Default)

@Test
fun `test flow launch`() = runBlocking {
    val job = events()
        .onEach { event -> println("event:$event") }
        .launchIn(CoroutineScope(Dispatchers.IO))
    delay(200)
    job.cancelAndJoin()
}

launchIn需要传入协程作用域，返回的是Job，这样可以方便的取消停止流

流的取消

流采用与协程同样的协作取消。流的收集可以是当流在一个可取消的挂起函数（例如delay）中挂起的时候取消

代码语言：javascript复制
fun simpleFlow3() = flow<Int> { 
    for (i in 1..3) {
        delay(1000)
        println(1000)
        emit(i)
    }
}
@Test
fun `test cancel flow`() = runBlocking {
   withTimeoutOrNull(2500){
       simpleFlow3().collect { value -> println(value) }
   }
    println("done")
}
流的取消检测
  • 为方便起见，流构建器对每个发射值执行附加的ensureActive检测以进行取消，这意味着从flow{}发出的繁忙循环是可以取消的
  • ensureActive检测的是协程job的状态，取消的话也是取消协程
  • 出于性能原因，大多数其他流操作不会自行执行其他取消检测，在协程出于繁忙循环的情况下，必须明确检测是否取消
  • 通过cancellable操作符来执行此操作
代码语言：javascript复制
fun simpleFlow3() = flow<Int> {
    for (i in 1..5) {
        delay(1000)
        println(1000)
        emit(i)
    }
}
//只会输出到3
@Test
fun `test cancel flow3`() = runBlocking {
    simpleFlow3().collect { value ->
        println(value)
        if (value == 3){
            cancel()
        }
    }
    //如果不加cancellable()就不能取消
    (1..5).asFlow().cancellable().collect { value ->
        println(value)
        if (value == 3){
            cancel()
        }
    }
}
背压与处理

背压：生产者效率大于消费者效率

  • buffer()，并发运行流中发射元素的代码。（也可以用flowOn切换上下文来实现）
  • conflate()，合并发射项，不对每个值进行处理，比如1-3，只处理1和3，中间的值不处理
  • collectLatest()，取消并重新发射最后一个值
  • 当必须更改CoroutineDispatcher时，flowOn操作符使用了相同的缓冲机制，但是buffer函数显示地请求缓冲而不改变执行上下文
代码语言：javascript复制
@Test
fun `test flow back pressure`() = runBlocking {
    val time = measureTimeMillis {
        simpleFlow3()
            .buffer(50)
            //.conflate()
            //.flowOn(Dispatchers.Default)
            .collect { value ->
                delay(300)
                println("collected $value")
            }
    }
}
操作符
过渡流操作符
  • 转换操作符：map、transform
  • 限长操作符：take
末端流操作符

末端操作符是在流上用于启动流收集的挂起函数，collect是最基本的末端操作符

  • 转化为各种集合，例如toList与toSet
  • 获取第一个（first）值与确保流发射单个（single）值的操作符
  • 使用reduce和fold将流规约到单个值
代码语言：javascript复制
@Test
fun `test flow operator`() = runBlocking<Unit> {
    val sum = (1..5).asFlow()
        .map { it * it }
        .reduce{a,b -> a b}
    //输出55
    println(sum)
}
组合操作符
  • 就像kotlin标准库中的sequence.zip扩展函数一样，流拥有一个zip操作符用于组合两个流中的相关值
  • 2个流是异步的
代码语言：javascript复制
@Test
fun `test flow zip`() = runBlocking<Unit> {
    val nums = (1..3).asFlow()
    val strs = flowOf("one", "two", "three")
    nums.zip(strs){ a,b -> "$a -> $b"}.collect { print(it) }
    //会输出1->one,2->two，3->three
}
展平流
  • flatMapConcat连接模式
  • flatMapMerge合并模式
  • flatMapLatest最新展平模式
流的异常处理

当运算符中的发射器或代码抛出异常时，处理方法：

  • try/catch块，捕获下游collect异常
  • catch函数（上游emit异常，可以在catch中恢复，比如补充数据）
代码语言：javascript复制
@Test
fun `test flow exception`() = runBlocking<Unit> {
    flow { 
        emit(1)
        throw ArithmeticException("Div 0")
    }.catch { e:Throwable ->
        println("catch $e")
        //补充元素
        emit(10)
    }.flowOn(Dispatchers.IO).collect{println(it)}
}
流的完成

当流收集完成时（普通情况或异常情况），它可能需要执行一个动作

  • 命令式finally块
  • onCompletion声明式处理，onCompletion还能拿到异常信息，但是不能捕获异常，同时能获取上游异常信息和下游异常信息
代码语言：javascript复制
@Test
fun `test flow exception onCompletion`() = runBlocking<Unit> {
    flow {
        emit(1)
        throw ArithmeticException("Div 0")
    }.onCompletion { exception -> 
        if (exception != null){
            println("flow completed exception: $exception")
        }}
        .catch { e:Throwable ->
        println("catch $e")
    }.collect{println(it)}
}

码字不易，求转发，求点在看，求关注，感谢！

android kotlin 函数 协程 异步

0 人点赞