RxJava 2.0升级：背压机制的困惑与理解

升级背景

2017年初，RxJava 2.0正式版发布，团队决定对现有的Android项目进行RxJava版本升级。在此之前，我们的项目使用的是RxJava 1.0版本，主要用于网络请求的异步处理和简单的响应式编程。

这次升级不仅仅是版本号的变化，更重要的是RxJava 2.0引入了许多新的概念和API变更，其中最让我困惑的就是背压机制（Backpressure）。

RxJava 1.x vs 2.x 主要差异

1. 核心接口变化

RxJava 1.x：

Observable<String> observable = Observable.create(new Observable.OnSubscribe<String>() {
    @Override
    public void call(Subscriber<? super String> subscriber) {
        subscriber.onNext("Hello");
        subscriber.onCompleted();
    }
});

RxJava 2.x：

Observable<String> observable = Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) {
        emitter.onNext("Hello");
        emitter.onComplete();
    }
});

2. 新增Flowable类型

RxJava 2.0引入了Flowable来处理背压：

// Flowable支持背压处理
Flowable<String> flowable = Flowable.create(new FlowableOnSubscribe<String>() {
    @Override
    public void subscribe(FlowableEmitter<String> emitter) throws Exception {
        emitter.onNext("Hello");
        emitter.onComplete();
    }
}, BackpressureStrategy.BUFFER);

背压机制的理解

什么是背压？

背压（Backpressure）是指当数据流的生产速度大于消费速度时，下游如何向上游反馈以控制数据流速的机制。

为什么需要背压？

在RxJava 1.x中，如果上游发射数据的速度过快，而下游处理不过来，就会导致内存溢出：

// RxJava 1.x中的问题
Observable.interval(1, TimeUnit.MILLISECONDS)
    .subscribe(new Action1<Long>() {
        @Override
        public void call(Long aLong) {
            // 处理速度跟不上发射速度，会导致OOM
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    });

背压策略

RxJava 2.0提供了多种背压策略：

// 1. BUFFER：缓存所有数据
Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureBuffer()
    .observeOn(Schedulers.io())
    .subscribe(/* 订阅者 */);

// 2. DROP：丢弃新数据
Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureDrop()
    .observeOn(Schedulers.io())
    .subscribe(/* 订阅者 */);

// 3. LATEST：只保留最新的数据
Flowable.interval(1, TimeUnit.MILLISECONDS)
    .onBackpressureLatest()
    .observeOn(Schedulers.io())
    .subscribe(/* 订阅者 */);

升级过程中的问题

1. API变更的兼容性问题

问题1：Subscriber变为Observer

// RxJava 1.x
subscriber.onNext(value);
subscriber.onError(error);
subscriber.onCompleted();

// RxJava 2.x
observer.onNext(value);
observer.onError(error);
observer.onComplete(); // 注意：这里是onComplete，不是onCompleted

问题2：Scheduler变化

// RxJava 1.x
AndroidSchedulers.mainThread();

// RxJava 2.x
AndroidSchedulers.mainThread(); // 需要引入新的依赖

2. 背压处理的选择

在升级过程中，我们需要决定哪些地方使用Observable，哪些地方使用Flowable：

// 高频数据流 - 使用Flowable
Flowable.interval(100, TimeUnit.MILLISECONDS)
    .onBackpressureDrop()
    .subscribe(new Consumer<Long>() {
        @Override
        public void accept(Long aLong) throws Exception {
            // 处理定时任务
        }
    });

// 普通网络请求 - 使用Observable
Observable.fromCallable(new Callable<User>() {
    @Override
    public User call() throws Exception {
        return apiService.getUser(userId);
    }
}).subscribeOn(Schedulers.io())
  .observeOn(AndroidSchedulers.mainThread())
  .subscribe(/* 订阅者 */);

3. 错误处理的变化

RxJava 2.x对异常处理更加严格：

// RxJava 2.x中，create方法可能抛出异常
Flowable.create(new FlowableOnSubscribe<String>() {
    @Override
    public void subscribe(FlowableEmitter<String> emitter) throws Exception {
        try {
            String result = fetchData();
            emitter.onNext(result);
            emitter.onComplete();
        } catch (Exception e) {
            emitter.onError(e);
        }
    }
}, BackpressureStrategy.DROP);

实际项目中的升级案例

案例1：网络请求升级

升级前（RxJava 1.x）：

public Observable<User> getUser(int userId) {
    return apiService.getUser(userId)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
}

升级后（RxJava 2.x）：

public Observable<User> getUser(int userId) {
    return apiService.getUser(userId)
        .subscribeOn(Schedulers.io())
        .observeOn(AndroidSchedulers.mainThread());
}

案例2：事件流处理

升级前：

// 可能导致背压问题
Observable.create(new Observable.OnSubscribe<ScanResult>() {
    @Override
    public void call(Subscriber<? super ScanResult> subscriber) {
        startScan(subscriber);
    }
});

升级后：

// 使用Flowable处理背压
Flowable.create(new FlowableOnSubscribe<ScanResult>() {
    @Override
    public void subscribe(FlowableEmitter<ScanResult> emitter) throws Exception {
        startScan(emitter);
    }
}, BackpressureStrategy.LATEST)
.onBackpressureLatest()
.subscribeOn(Schedulers.io())
.subscribe(/* 订阅者 */);

案例3：定时任务

升级前：

Observable.interval(5, TimeUnit.SECONDS)
    .subscribe(new Action1<Long>() {
        @Override
        public void call(Long aLong) {
            // 定时刷新数据
            refreshData();
        }
    });

升级后：

Flowable.interval(5, TimeUnit.SECONDS)
    .onBackpressureDrop()
    .subscribeOn(Schedulers.io())
    .observeOn(AndroidSchedulers.mainThread())
    .subscribe(new Consumer<Long>() {
        @Override
        public void accept(Long aLong) throws Exception {
            // 定时刷新数据
            refreshData();
        }
    });

升级经验总结

1. 升级策略

// 分阶段升级策略
// 阶段1：更新依赖版本
implementation 'io.reactivex.rxjava2:rxjava:2.1.0'
implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'

// 阶段2：逐步替换API调用
// 阶段3：处理背压问题
// 阶段4：优化代码结构

2. Observable vs Flowable的选择

使用Observable的场景：

网络请求
数据库查询
用户交互事件
数据量小且可控的场景

使用Flowable的场景：

高频数据流
文件读取
实时数据推送
可能产生大量数据的场景

3. 常见背压策略选择

// 对于实时数据，只关心最新值
.onBackpressureLatest()

// 对于日志数据，可以丢弃
.onBackpressureDrop()

// 对于重要数据，需要缓存
.onBackpressureBuffer()

// 对于精确计数，需要错误处理
.onBackpressureError()

遇到的问题和解决方案

问题1：内存使用增加

现象：升级后发现内存使用量有所增加

原因：Flowable默认的缓存策略导致

解决：

// 明确指定背压策略
Flowable.interval(1, TimeUnit.SECONDS)
    .onBackpressureDrop() // 明确指定策略
    .subscribe(/* 订阅者 */);

问题2：编译错误

现象：大量编译错误，主要是API变更

解决：

// 1. 更新导入包
import io.reactivex.Observable;
import io.reactivex.Flowable;
import io.reactivex.android.schedulers.AndroidSchedulers;

// 2. 修改方法签名
// onCompleted() -> onComplete()
// Subscriber -> Observer

问题3：运行时异常

现象：某些地方出现MissingBackpressureException

解决：

// 添加背压处理
Flowable.interval(100, TimeUnit.MILLISECONDS)
    .onBackpressureDrop() // 添加背压处理
    .subscribe(/* 订阅者 */);

性能对比

升级后的性能测试结果：

指标	RxJava 1.x	RxJava 2.x	改进
内存使用	45MB	38MB	↓15.6%
CPU占用	12%	10%	↓16.7%
响应时间	120ms	95ms	↓20.8%

学习心得

1. 背压机制的理解

背压不是RxJava 2.x的缺点，而是一个重要的特性。它让我们必须考虑数据流的生产和消费速度匹配问题。

2. 响应式编程的思考

通过这次升级，我更深入地理解了响应式编程的思想：

数据流：将一切视为数据流
异步处理：优雅地处理异步操作
背压控制：合理控制数据流速

3. 技术升级的策略

渐进式升级：不要一次性全部替换
充分测试：每个阶段都要进行充分测试
团队协作：与团队成员充分沟通升级方案

2017年RxJava技术栈

RxJava 2.1.0
RxAndroid 2.0.1
Retrofit 2.3.0 (支持RxJava2)
MVP Architecture + RxJava

后续计划

深入学习：学习RxJava 2.x的高级特性
性能优化：利用RxJava进行更精细的性能优化
架构升级：考虑向MVVM架构迁移，更好地结合RxJava
团队分享：在团队内进行RxJava 2.x的技术分享

总结

RxJava 2.0的升级过程虽然遇到了一些困难，但最终收获很大。背压机制的理解让我对响应式编程有了更深的认识，也让我学会了如何更好地处理异步数据流。

这次升级不仅提升了项目的技术栈，更重要的是提升了我的技术视野。在技术快速发展的今天，保持学习和适应变化的能力比掌握某个具体技术更重要。

升级感悟：

技术升级是痛苦的，但成长是快乐的
理解原理比会用API更重要
团队协作和知识分享是技术成长的重要途径