RxJava1. 理解核心概念与基础操作1.1 核心四要素 (The Observable Contract)1.2 冷 Observable 与 热 Observable1.3 基础操作符 (Operators)1.3.1 创建操作符 (Creating):1.3.2 变换操作符 (Transforming):1.3.3 过滤操作符 (Filtering):1.3.4 组合操作符 (Combining):1.4. 线程调度 (Schedulers)2. 深入原理与高级操作2.1 操作符链与 Lift 原理2.2 背压 (Backpressure) - RxJava 2/3 的核心概念2.3. 高级操作符2.4. Subject (主题)2.5. 整体变换 (compose 和 transform)1. 理解核心概念与基础操作这是学习 RxJava 的基石必须牢固掌握。1.1 核心四要素 (The Observable Contract)RxJava 的核心是 观察者模式 (Observer Pattern)它定义了四个关键角色•Observable(被观察者/数据源): 产生数据流的源头。可以发射 多个数据项以成功或失败结束。•Observer(观察者/订阅者): 接收并处理 Observable 发射的数据。它定义了四个方法• onSubscribe(Disposable d): 订阅建立时调用可用于取消订阅。• onNext(T value):接收到一个数据项时调用。• onError(Throwable e): 发生错误时调用流终止。• onComplete():流成功完成时调用流终止。•Subscription(订阅): 连接 Observable 和 Observer 的纽带。通过调用 subscribe() 方法建立。•Disposable(可dispose的): 代表一个订阅关系调用其 dispose() 方法可以取消订阅停止接收 数据并释放资源。Hello World 示例:import io.reactivex.rxjava3.core.Observable; import io.reactivex.rxjava3.disposables.Disposable; public class RxJavaHelloWorld { public static void main(String[] args) { // 1. 创建 Observable (数据源) Observable String observable Observable.just(Hello, RxJava!); // 2. 创建 Observer (观察者) MyObserver observer new MyObserver(); // 3. 建立订阅 (Subscription) Disposable disposable observable.subscribe(observer); // 4. (可选) 取消订阅 // disposable.dispose(); } static class MyObserver implements io.reactivex.rxjava3.core.Observer String { Override public void onSubscribe(io.reactivex.rxjava3.disposables.Disposable d) { System.out.println(订阅建立); } Override public void onNext(String s) { System.out.println(接收到: s); } Override public void onError(Throwable e) { System.err.println(发生错误: e.getMessage()); } Override public void onComplete() { System.out.println(流已结束); } } } // 输出: // 订阅建立 // 接收到: Hello // 接收到: RxJava! // 流已结束1.2 冷 Observable 与 热 Observable• 冷 Observable (Cold Observable): 只有在有订阅者订阅时才开始发射数据。每个订阅者都会收到完整的数据流。just, fromArray, create 等创建的通常是冷的。• 热 Observable (Hot Observable): 无论是否有订阅者都会发射数据。订阅者只能接收到订阅之后发射的数据。PublishSubject, ReplaySubject 等创建的是热的。1.3 基础操作符 (Operators)操作符是 RxJava 的灵魂用于对数据流进行转换、过滤、组合等。1.3.1 创建操作符 (Creating):• just(T…): 发射指定的项。• fromArray(T[]) / fromIterable(Iterable? extends T): 从数组或集合发射数据。• create(ObservableOnSubscribe): 通过自定义逻辑创建 Observable。• interval(long, TimeUnit): 定时发射递增的数字。• range(int start, int count): 发射一个范围内的整数序列。1.3.2 变换操作符 (Transforming):• map(FunctionT, R): 将每个发射的项通过一个函数转换成另一个值。Observable.just(1, 2, 3 ) .map(Integer::parseInt) // 将 String 转为 Integer .subscribe(System.out::println); // 输出: 1, 2, 3• flatMap(FunctionT, ObservableSource? extends R): 将每个发射的项转换成一个新的 Observable然后将这些 Observable 发射的数据“拍平”并合并到一个单一的 Observable 中。常用于异步操作。Observable.just(url1, url2) .flatMap(url - fetchDataFromNetwork(url)) // 假设 fetchDataFromNetwork 返回 ObservableData .subscribe(data - System.out.println(Received: data));• concatMap(FunctionT, ObservableSource? extends R): 类似 flatMap但保证内部 Observable 的发射顺序与原始项的顺序一致。1.3.3 过滤操作符 (Filtering):• filter(Predicate): 只发射满足指定条件的项。Observable.just(1, 2, 3, 4, 5) .filter(x - x % 2 0) // 只保留偶数 .subscribe(System.out::println); // 输出: 2, 4• take(long count): 只取前 N 项。• skip(long count): 跳过前 N 项。1.3.4 组合操作符 (Combining):• merge(ObservableSource? extends T…): 将多个 Observable 发射的数据合并不保证顺序。• concat(ObservableSource? extends T…): 按顺序连接多个 Observable前一个完成后才订阅下一个。• zip(ObservableSource, ObservableSource, BiFunctionT1, T2, R): 将多个 Observable 的数据按顺序一一配对组合。ObservableString names Observable.just(Alice, Bob); ObservableInteger ages Observable.just(25, 30); Observable.zip(names, ages, (name, age) - name is age years old.) .subscribe(System.out::println); // 输出: // Alice is 25 years old. // Bob is 30 years old.• combineLatest(ObservableSource, ObservableSource, BiFunctionT1, T2, R): 当任意一个 Observable 发射新数据时与其它 Observable 的最新数据组合。1.4. 线程调度 (Schedulers)RxJava 强大的异步能力离不开调度器。• subscribeOn(Scheduler scheduler): 指定 上游数据产生、操作符执行在哪个线程执行。只对第一个 subscribeOn有效。• observeOn(Scheduler scheduler): 指定下游onNext, onError, onComplete在哪个线程执行。可以多次调用,切换线程。常用调度器:• Schedulers.io(): 用于 I/O 密集型操作如网络请求、文件读写。• Schedulers.computation(): 用于 CPU 密集型计算。• Schedulers.newThread(): 为每个任务创建一个新线程。• AndroidSchedulers.mainThread() (RxAndroid): 在 Android 主线程执行用于更新 UI。示例:Observable.just(url) .subscribeOn(Schedulers.io()) // 网络请求在 IO 线程 .map(url - fetchData(url)) // 转换操作也在 IO 线程 .observeOn(AndroidSchedulers.mainThread()) // 切换到主线程更新 UI .subscribe(data - updateUI(data));2. 深入原理与高级操作掌握了基础后需要理解其内部机制和更复杂的操作。2.1 操作符链与 Lift 原理理解操作符是如何串联起来的至关重要。每个操作符如 map, filter内部通常会调用 lift(Operator) 方法创建一个新的 Observable这个新的 Observable 会包装上游的 Observable并在订阅时将下游的 Observer 包装成一个新的 Observer从而在数据传递过程中插入转换或过滤逻辑。这是一个典型的装饰者模式。2.2 背压 (Backpressure) - RxJava 2/3 的核心概念在异步场景下如果生产者Observable发射数据的速度远快于消费者Observer处理数据的速度会导致内存溢出。RxJava 2 引入了 Flowable 来专门处理背压问题。• Flowable vs Observable:• Observable: 无背压支持适用于 GUI 事件、短序列等。• Flowable: 有背压支持适用于网络请求、数据库操作、文件 I/O 等可能产生大量数据的场景。• 背压策略 (BackpressureStrategy):• BUFFER: 缓存所有数据可能导致 OOM。• DROP: 丢弃无法处理的数据。• LATEST: 只保留最新的数据丢弃旧的。• ERROR: 抛出 MissingBackpressureException。• MISSING: 不做任何处理需要手动通过 onBackpressureXXX 操作符处理。2.3. 高级操作符• switchMap: 类似 flatMap但当源 Observable 发射一个新项时会取消订阅并丢弃前一个内部 Observable 产生的数据只处理最新的内部 Observable。• debounce: 如果在一个指定的时间段内没有新的数据发射则发射最近的一个数据。常用于搜索框防抖。• distinct / distinctUntilChanged: 过滤掉重复的数据。• retry / retryWhen: 在发生错误时重试。• timeout: 如果在指定时间内没有收到数据则抛出超时异常。• doOnNext / doOnError / doOnComplete / doFinally: 用于副作用side-effect如日志记录、资源清理不影响数据流本身。2.4. Subject (主题)Subject 既是 Observable 又是 Observer可以手动向数据流中“推送”数据是创建热 Observable 的主要方式。• PublishSubject: 向所有订阅者广播数据订阅后才能收到数据。• ReplaySubject: 缓存所有发射过的数据新的订阅者会收到所有历史数据。• BehaviorSubject: 缓存最后一个数据新的订阅者会立即收到这个最新数据然后接收后续数据。• AsyncSubject: 只在流完成时向所有订阅者发射最后一个数据。BehaviorSubject 示例 (模拟状态管理):BehaviorSubjectString userNameSubject BehaviorSubject.createDefault(Guest); // 订阅者 A userNameSubject.subscribe(name - System.out.println(User A sees: name)); // 输出: User A sees: Guest // 订阅者 B (稍后订阅) userNameSubject.subscribe(name - System.out.println(User B sees: name)); // 输出: User B sees: Guest // 更新用户名 userNameSubject.onNext(Alice); // 输出: User A sees: Alice // 输出: User B sees: Alice2.5. 整体变换 (compose 和 transform)当一组操作符需要在多个地方复用时可以使用 compose 或 transform。• compose(ObservableTransformerT, R): 用于 Observable。• to(ObservableConverterT, R): 更通用的转换。示例- 封装网络请求的通用线程切换:public class SchedulersTransformerT implements ObservableTransformerT, T { Override public ObservableSourceT apply(ObservableT upstream) { return upstream.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()); } } // 使用 apiService.getData() .compose(new SchedulersTransformer()) // 应用通用变换 .subscribe(...);