rxjava怎么处理背压

在 RxJava 中，背压（Backpressure） 指的是：
上游发射数据的速度 > 下游处理数据的速度 时，如何防止 OOM 或系统崩溃。

RxJava 主要通过 Flowable + 背压策略 来处理，而 Observable 默认不支持背压（会直接抛 MissingBackpressureException）。

下面按「核心概念 → 处理方式 → 示例」来说明。

一、Observable vs Flowable

Observable.create(...)   // 无背压
Flowable.create(...)     // 有背压

二、Flowable 的背压策略（BackpressureStrategy）

在创建 Flowable 时指定：

Flowable.create(emitter -> {
    for (int i = 0; i < 1000; i++) {
        emitter.onNext(i);
    }
    emitter.onComplete();
}, BackpressureStrategy.BUFFER)

常见策略

三、下游主动拉取数据（request）

Flowable 是 响应式拉取模型：

Flowable.range(1, 100)
    .onBackpressureDrop()
    .subscribe(new Subscriber<Integer>() {

        private Subscription subscription;

        @Override
        public void onSubscribe(Subscription s) {
            subscription = s;
            s.request(1); // 请求 1 个
        }

        @Override
        public void onNext(Integer integer) {
            System.out.println(integer);
            subscription.request(1); // 处理完再要
        }

        @Override
        public void onError(Throwable t) {}

        @Override
        public void onComplete() {}
    });

✅ 这是最安全、推荐的方式

四、操作符层面的背压处理

1️⃣ 缓冲（buffer / window）

Flowable.interval(1, TimeUnit.MILLISECONDS)
    .buffer(100)
    .subscribe(list -> System.out.println(list.size()));

2️⃣ 限流（sample / throttle）

Flowable.interval(10, TimeUnit.MILLISECONDS)
    .sample(1, TimeUnit.SECONDS)
    .subscribe(System.out::println);

3️⃣ 丢弃或取最新

.onBackpressureDrop()
.onBackpressureLatest()

五、Observable 转 Flowable（不推荐但可用）

Observable.range(1, 100)
    .toFlowable(BackpressureStrategy.BUFFER);

⚠️ 只是“强制兼容”，并不能真正解决背压

六、最佳实践总结

✅ 高频数据流 → Flowable
✅ 下游 request 控制消费速度
✅ 避免 BUFFER 无限增长
✅ UI 事件用 Observable，数据流用 Flowable

如果你愿意，我可以：

• 用 真实业务场景（网络 + 数据库）举例
• 对比 RxJava 2 / 3 背压差异
• 讲 为什么 Observable 不支持背压

你更想看哪一种？