做教育的网站智慧教育

一、什么是背压（Backpressure）？

背压是指：数据生产速度 > 数据消费速度，导致的数据堆积问题。

举个简单例子：

假设你在做一个网络数据流展示的功能：

• 数据源：每秒从服务器推送 1000 条日志。

• 你的处理逻辑：只能每秒显示/处理 100 条。

这时，900 条数据每秒堆积下来。如果不处理，几秒后内存就爆炸了。

---

二、背压问题的根本来源

1. 不对称的速率

• 生产者异步、高速、不断产出数据。

• 消费者是有限资源（UI线程、IO线程、磁盘、数据库等）。

2. 异步流处理天然存在“流速失控”的可能

• 你用的流式框架（RxJava、Flow）默认都很“热情”，默认生产者一口气发完。

• 如果没有对消费者能力加限制，就出现“背压”问题。

---

三、背压未处理的后果

• OOM（内存溢出）：缓存数据太多。

• ANR（卡死）：主线程被大量任务挤压。

• UI 卡顿/掉帧：不断处理 backlog。

• 数据丢失：部分框架默认丢掉旧数据（比如 ConflatedChannel）。

• 崩溃：RxJava 默认会抛出 MissingBackpressureException。

---

四、背压处理的核心策略（框架无关）

策略	含义	举例
缓存（Buffer）	用队列先缓存，慢慢消费	Flow 的 buffer()
丢弃（Drop）	不处理旧数据，保活系统	Flow 的 conflate()，Rx 的 onBackpressureDrop()
保留最新（Latest）	总是处理最新值	Flow 的 collectLatest()，Rx 的 onBackpressureLatest()
限速（Throttle）	控制发射频率	Rx 的 throttleFirst()，Flow 的 debounce()
反压反馈（Request）	消费者主动拉取	Rx 的 Flowable.request()
挂起/阻塞	不缓存，发射阻塞等待消费	Flow 默认行为

---

五、完整示例对比：RxJava vs Kotlin Flow 背压处理

---

✅ Kotlin Flow 背压处理示例

kotlin

import kotlinx.coroutines.*

import kotlinx.coroutines.flow.*

fun main() = runBlocking {

                 println("Start")

                val flow = flow {

                        for (i in 1..1000) {

                                println("Emit: $i") emit(i)

                        }

                }

                .buffer(capacity = 50) // 缓冲最多50条

                //.conflate() // 或者只保留最新

                //.debounce(100) // 或控制速率

                flow.collect {

                        value -> delay(100) // 模拟处理耗时                         println("Collect: $value")

                 }

                  println("End")

}

👉 输出节选：

Emit: 1

Emit: 2 ...

Collect: 1

Collect: 2 ...

你可以通过 .buffer()、.conflate()、.collectLatest() 等方式切换不同策略。

---

✅ RxJava Flowable 背压处理示例

java

复制编辑

Flowable.interval(1, TimeUnit.MILLISECONDS) // 每毫秒发一次 .onBackpressureDrop() // 丢弃处理不了的

.observeOn(Schedulers.io()) // 切换线程处理

.subscribe(item -> {

        Thread.sleep(100); // 模拟慢处理

        System.out.println("Received: " + item);

}, Throwable::printStackTrace);

👉 输出节选：

Received: 5

Received: 102

Received: 201 ...

你会发现中间的很多值被“跳过”，因为你每100ms才能处理一个，而上游每1ms就发一个，系统只能保最新。

---

六、总结建议（Android 实战）

场景	建议方案
UI 层收集数据（如 LiveData -> Compose）	用 Flow，buffer() + collectLatest()
高频传感器或点击事件	使用 debounce() 或 throttleFirst()
网络请求分页/长轮询	Flow + Retrofit + emitAll()
文件读写、数据库导入	Flowable/Flow + onBackpressureBuffer()
多生产者合并数据	使用 merge() + 合理背压策略