本站首页 收藏本站
广告位联系

APP正在开发中...
返回顶部
您的位置： 主页 > 软件编程 > Android >

收藏
分享到

Flow解决背压问题的方法介绍

Android 来源：互联网 作者：佚名 发布时间：2022-11-15 16:30:44 人浏览
摘要

随着时间的推移，越来越多的主流应用已经开始全面拥抱Kotlin，协程的引入，Flow的诞生，给予了开发很多便捷，作为协程与响应式编程结合的流式处理框架，一方面它简单的数据转换与

随着时间的推移，越来越多的主流应用已经开始全面拥抱Kotlin，协程的引入，Flow的诞生，给予了开发很多便捷，作为协程与响应式编程结合的流式处理框架，一方面它简单的数据转换与操作符，没有繁琐的操作符处理，广受大部分开发的青睐，另一方面它并没有响应式编程带来的背压问题（BackPressure）的困扰；接下来，本文将会就Flow如何解决背压问题进行探讨

关于背压（BackPressure）

背压问题是什么

首先我们要明确背压问题是什么，它是如何产生的？简单来说，在一般的流处理框架中，消息的接收处理速度跟不上消息的发送速度，从而导致数据不匹配，造成积压。如果不及时正确处理背压问题，会导致一些严重的问题

  • 比如说，消息拥堵了，系统运行不畅从而导致崩溃
  • 比如说，资源被消耗殆尽，甚至会发生数据丢失的情况

如下图所示，可以直观了解背压问题的产生,它在生产者的生产速率高于消费者的处理速率的情况下出现

定义背压策略

既然知道了背压问题我们已经知道是如何产生的，就要去尝试如何正确处理它，大致意思是，如果你有一个流，你需要一个缓冲区，以防数据产生的速度快于消耗的速度，所以往往就会针对这个背压策略进行些讨论

  • 定义的中间缓冲区需要多大才比较合适？
  • 如果缓冲区数据已满了，我们怎么样处理新的事件？

对于以上问题，通过学习Flow里的背压策略，相信可以很快就知道答案了

Flow的背压机制

由于Flow是基于协程中使用的，它不需要一些巧妙设计的解决方案来明确处理背压，在Flow中，不同于一些传统的响应式框架，它的背压管理是使用Kotlin挂起函数suspend实现的，看下源码你会发现，它里面所有的函数方法都是使用suspend修饰符标记，这个修饰符就是为了暂停调度者的执行不阻塞线程。因此，Flow<T>在同一个协程中发射和收集时，如果收集器跟不上数据流，它可以简单地暂停元素的发射，直到它准备好接收更多。看到这，是不是觉得有点难懂.......

简单举个例子，假设我们拥有一个烤箱，可以用来烤面包，由于烤箱容量的限制，一次只能烤4个面包，如果你试着一次烤8个面包，会大大加大烤箱的承载负荷，这已经远远超过了它的内存使用量，很有可能会因此烧掉你的面包。

模拟背压问题

回顾下之前所说的，当我们消耗的速度比生产的速度慢的时候，就会产生背压，下面用代码来模拟下这个过程

首先先创建一个方法，用来每秒发送元素

1

2

3

4

5

6

7

8

9

10

11

12

fun currentTime() = System.currentTimeMillis()

fun threadName() = Thread.currentThread().name

var start: Long = 0

?

fun createEmitter(): Flow<Int> =

    (1..5)

        .asFlow()

        .onStart { start = currentTime() }

        .onEach {

            delay(1000L)

            print("Emit $it (${currentTime() - start}ms) ")

        }

接着需要收集元素,这里我们延迟3秒再接收元素, 延迟是为了夸大缓慢的消费者并创建一个超级慢的收集器。

1

2

3

4

5

6

7

8

9

10

11

12

fun main() {

    runBlocking {

        val time = measureTimeMillis {

            createEmitter().collect {

                print("\nCollect $it starts ${start - currentTime()}ms")

                delay(3000L)

                println("   Collect $it ends ${currentTime() - start}ms")

            }

        }

        print("\nCollected in $time ms")

    }

}

看下输出结果，如下图所示

这样整个过程下来，大概需要20多秒才能结束，这里我们模拟了接收元素比发送元素慢的情况，因此就需要一个背压机制，而这正是Flow本质中的，它并不需要另外的设计来解决背压

背压处理方式

使用buffer进行缓存收集

为了使缓冲和背压处理正常工作，我们需要在单独的协程中运行收集器。这就是.buffer()操作符进来的地方，它是将所有发出的项目发送Channel到在单独的协程中运行的收集器。

1

2

3

4

public fun <T> Flow<T>.buffer(

    capacity: Int = BUFFERED,

    onBufferOverflow: BufferOverflow = BufferOverflow.SUSPEND

): Flow<T>

它还为我们提供了缓冲功能，我们可以指定capacity我们的缓冲区和处理策略onBufferOverflow，所以当Buffer溢出的时候，它为我们提供了三个选项

1

2

3

4

5

enum BufferOverflow {

   SUSPEND，

   DROP_OLDEST，

   DROP_LATEST

 }
  • 默认使用SUSPEND：会将当前协程挂起，直到缓冲区中的数据被消费了
  • DROP_OLDEST：它会丢弃最老的数据
  • DROP_LATEST: 它会丢弃最新的数据

好的，我们回到上文所展示的模拟示例，这时候我们可以加入缓冲收集buffer，不指定任何参数，这样默认就是使用SUSPEND，它会将当前协程进行挂起

此时当收集器繁忙的时候，程序就开始缓冲，并在第一次收集方法调用结束的时候，两次发射后再次开始收集，此时流程的耗时时长缩短到大约16秒就可以执行完毕，如下图所示输出结果

使用conflate解决

conflate操作符于Channel中的Conflate模式是一直的，新数据会直接覆盖掉旧数据，它不设缓冲区，也就是缓冲区大小为 0，丢弃旧数据，也就是采取 DROP_OLDEST 策略，那么不就等于buffer(0,BufferOverflow.DROP_OLDEST)，可以看下它的源码可以佐证我们的判断

1

public fun <T> Flow<T>.conflate(): Flow<T> = buffer(CONFLATED)

在某些情况下，由于根本原因是解决生产消费速率不匹配的问题，我们需要做一些取舍的操作，conflate将丢弃掉旧数据，只有在收集器空闲之前发出的最后一个元素才被收集，将上文的模拟实例改为conflate执行，你会发现我们直接丢弃掉了2和4，或者说新的数据直接覆盖掉了它们，整个流程只需要10秒左右就执行完成了

使用collectLatest解决

通过官方介绍，我们知道collectLatest作用在于当原始流发出一个新的值的时候，前一个值的处理将被取消，也就是不会被接收， 和conflate的区别在于它不会用新的数据覆盖，而是每一个都会被处理，只不过如果前一个还没被处理完后一个就来了的话，处理前一个数据的逻辑就会被取消

1

suspend fun <T> Flow<T>.collectLatest(action: suspend (T) -> Unit)

还是上文的模拟实例，这里我们使用collectLatest看下输出结果：

这样也是有副作用的，如果每个更新都非常重要，例如一些视图，状态刷新，这个时候就不必要用collectLatest; 当然如果有些更新可以无损失的覆盖，例如数据库刷新，就可以使用到collectLatest，具体详细的使用场景，还需要靠开发者自己去衡量选择使用

小结

对于Flow可以说不需要额外提供什么巧妙的方式解决背压问题，Flow的本质，亦或者说Kotlin协程本身就已经提供了相应的解决方案；开发者只需要在不同的场景中选择正确的背压策略即可。总的来说，它们都是通过使用Kotlin挂起函数suspend，当流的收集器不堪重负时，它可以简单地暂停发射器，然后在准备好接受更多元素时恢复它。

关于挂起函数suspend这里就不过多赘述了，只需要明白的一点是它与传统的基于线程的同步数据管道中背压管理非常相似，无非就是，缓慢的消费者通过阻塞生产者的线程自动向生产者施加背压，简单来说，suspend通过透明地管理跨线程的背压而不阻塞它们，将其超越单个线程并进入异步编程领域。
版权声明 : 本文内容来源于互联网或用户自行发布贡献，该文观点仅代表原作者本人。本站仅提供信息存储空间服务和不拥有所有权，不承担相关法律责任。如发现本站有涉嫌抄袭侵权， 违法违规的内容， 请发送邮件至2530232025#qq.cn(#换@)举报，一经查实，本站将立刻删除。

您可能感兴趣的文章 :

原文链接 : https://juejin.cn/post/7165380647304282126
    Tag :
相关文章
  • Android硬件解码组件MediaCodec使用教程
    1.MediaCodec 是什么 MediaCodec类可以访问底层媒体编解码器框架（StageFright 或 OpenMAX），即编解码组件。是Android 的低层多媒体基础设施的一部分

  • Flow解决背压问题的方法介绍

    Flow解决背压问题的方法介绍
    随着时间的推移，越来越多的主流应用已经开始全面拥抱Kotlin，协程的引入，Flow的诞生，给予了开发很多便捷，作为协程与响应式编程结合
  • Andorid开发中反射机制的介绍
    在andorid开发中，经常遇见在某些工具类中没有Context上下文对象时，一些系统服务的代理对象无法创建出来，举个例子：比如在源码(framewo

  • Flutter加载图片的多样玩法汇总

    Flutter加载图片的多样玩法汇总
    加载本地图片 在项目目录下创建assets文件夹，再在其文件夹下创建images文件夹，后面将需要的图片复制到其中即可 在pubspec.yaml文件中添加引

  • Kotlin的Collection与Sequence操作异同点介绍

    Kotlin的Collection与Sequence操作异同点介绍
    在Android开发中，集合是我们必备的容器，Kotlin的标准库中提供了很多处理集合的方法，而且还提供了两种基于容器的工作方式：Collection 和

  • 实现一个Kotlin函数类型方法

    实现一个Kotlin函数类型方法
    接口与函数类型 业务开发中，经常会有实现一个函数式接口（即接口只有一个方法需要实现）的场景，大家应该都会不假思索的写出如下代
  • Android10 App启动Activity源码分析
    ActivityThread的main方法 让我们把目光聚焦到ActivityThread的main方法上。 ActivityThread的源码路径为/frameworks/base/core/java/android/app/ActivityThread。 1 2

  • Android10客户端事务管理ClientLifecycleManager源码解析

    Android10客户端事务管理ClientLifecycleManager源码解析
    在Android 10 App启动分析之Activity启动篇（二）一文中，简单地介绍了Activity的生命周期管理器是如何调度Activity进入onCreate生命周期的流程。这

  • Kotlin对象的懒加载方式by lazy与lateinit异同介绍

    Kotlin对象的懒加载方式by lazy与lateinit异同介绍
    属性或对象的延时加载是我们相当常用的，一般我们都是使用 lateinit 和 by lazy 来实现。 他们两者都是延时初始化，那么在使用时那么他们两

  • Android类加载流程分析

    Android类加载流程分析
    本文分析的代码基于Android8.1.0源码。 流程分析 从loadClass开始，我们来看下Android中类加载的流程 /libcore/ojluni/src/main/java/java/lang/ClassLoader.ja
推荐阅读
最新更新
友情链接
  • 本站所有内容来源于互联网或用户自行发布，本站仅提供信息存储空间服务，不拥有版权，不承担法律责任。如有侵犯您的权益，请您联系站长处理！
  • Copyright © 2017-2022 F11.CN All Rights Reserved. F11站长开发者网 版权所有 | 苏ICP备2022031554号-1 | 51LA统计