RxJava 核心原理：响应式流与操作符源码解析

在 Android 开发中，异步操作和事件处理无处不在。从传统的回调接口（Callback）到 Handler/Looper，再到 AsyncTask，开发者一直在寻找一种更优雅的方式来处理复杂的异步数据流。RxJava 凭借其强大的响应式编程模型和丰富的操作符，成为了处理异步任务的利器。

然而，仅仅会调用 map、flatMap 并不足以写出健壮的代码。RxJava 的学习曲线陡峭，其底层实现往往让初学者感到魔幻。本文将剥开 RxJava 的 API 外衣，深入探究其作为“响应式流”的底层运行机制。

什么是 RxJava？

一句话概括：RxJava 是一个基于事件流、实现异步操作的库。

它将一切抽象为“数据流”（Stream），并使用扩展的观察者模式来进行事件的生产与消费。

为什么需要 RxJava？

在复杂的业务场景中，我们经常遇到这样的情况：

1. 请求 A 成功后，拿着 A 的结果去请求 B。
2. 同时发起请求 C 和 D，等它们都返回后，将结果合并刷新 UI。
3. 在处理这些请求的过程中，随时可能发生异常，需要统一的错误处理。

如果使用传统的 Callback，代码会迅速陷入“回调地狱”（Callback Hell），形成横向发展的嵌套结构，难以阅读和维护。RxJava 通过链式调用（Fluent API）和极其丰富的操作符，将复杂的异步逻辑拉平成线性的代码结构。

核心隐喻：工业流水线

理解 RxJava 最贴切的比喻是工业生产的流水线：

• Observable（被观察者）：流水线的起点，负责源源不断地生产原材料（发送事件）。
• Observer（观察者）：流水线的终点，负责接收包装好的最终产品并进行消费。
• Operator（操作符）：流水线中间的一个个加工台，比如 map 负责将原材料上色，filter 负责剔除不合格的半成品。
• subscribe（订阅）：启动流水线的开关。在按下开关之前，流水线是静止的；一旦按下，原材料就会从起点顺着加工台一路流向终点。

核心骨架：扩展的观察者模式

RxJava 的核心骨架是观察者模式，但它比传统的观察者模式多了一些概念：

• onNext(T)：发送普通数据。
• onError(Throwable)：发送错误信号，一旦调用，数据流终止。
• onComplete()：发送完成信号，数据流正常结束。
• onSubscribe(Disposable)：建立订阅关系时的回调，可以通过 Disposable 切断流水线。

让我们从最基础的创建与订阅开始，看看底层发生了什么：

Observable.create(new ObservableOnSubscribe<String>() {
    @Override
    public void subscribe(ObservableEmitter<String> emitter) throws Exception {
        emitter.onNext("Hello");
        emitter.onComplete();
    }
}).subscribe(new Observer<String>() {
    @Override
    public void onSubscribe(Disposable d) { }

    @Override
    public void onNext(String s) {
        System.out.println(s);
    }

    @Override
    public void onError(Throwable e) { }

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

create 的底层机制

当你调用 Observable.create 时，实际上返回了一个 ObservableCreate 对象，它保存了你传入的 ObservableOnSubscribe（即生产数据的逻辑）。

// RxJava 源码简析
public static <T> Observable<T> create(ObservableOnSubscribe<T> source) {
    return RxJavaPlugins.onAssembly(new ObservableCreate<T>(source));
}

此时，数据并没有开始发送，这被称为冷流（Cold Stream）。

subscribe 的执行流

当调用 subscribe(Observer) 时，流水线启动。我们进入 Observable 的源码：

public final void subscribe(Observer<? super T> observer) {
    // 1. 包装 Observer
    observer = RxJavaPlugins.onSubscribe(this, observer);
    
    // 2. 核心：调用各个子类实现的 subscribeActual
    subscribeActual(observer);
}

在 ObservableCreate 的 subscribeActual 中：

@Override
protected void subscribeActual(Observer<? super T> observer) {
    // 1. 创建发射器，并将目标 Observer 传入
    CreateEmitter<T> parent = new CreateEmitter<T>(observer);
    
    // 2. 触发 onSubscribe 回调
    observer.onSubscribe(parent);
    
    try {
        // 3. 执行开发者定义的发射逻辑
        source.subscribe(parent);
    } catch (Throwable ex) {
        parent.onError(ex);
    }
}

发生了什么？ CreateEmitter 扮演了“中间人”的角色。开发者在自定义逻辑里调用的 emitter.onNext("Hello")，实际上是调用了 CreateEmitter.onNext。CreateEmitter 内部会做一些状态检查（是否已解除订阅、是否已抛出异常），然后将数据原封不动地传递给真正的 Observer.onNext("Hello")。

操作符的魔法：装饰器模式的极致应用

RxJava 的强大之处在于操作符。为什么 map 能在中间改变数据类型？为什么一条长长的链式调用能一层层处理数据？

其核心本质是：装饰器模式（Decorator Pattern）。

map 操作符的源码剖析

假设我们有一段代码：

Observable.create(...)   // 返回 ObservableCreate (记为 O1)
    .map(s -> s.length()) // 返回 ObservableMap (记为 O2)
    .subscribe(observer); // 传入真正的 Observer (记为 Obs1)

当我们调用 map 时：

public final <R> Observable<R> map(Function<? super T, ? extends R> mapper) {
    // 将当前的 Observable (O1) 和转换函数 mapper 包装到一个新的 ObservableMap 中
    return RxJavaPlugins.onAssembly(new ObservableMap<T, R>(this, mapper));
}

重点来了，当我们对最终返回的 O2 调用 subscribe(Obs1) 时，流程是这样的：

1. 自下而上的订阅构建： 调用 O2.subscribe(Obs1) -> O2 是 ObservableMap，其内部实现了 subscribeActual：

   @Override
   public void subscribeActual(Observer<? super U> t) {
       // O2 拿到下游的 Obs1，并将其包装为一个 MapObserver (记为 Obs2)
       // 然后把 Obs2 传给上游的 O1 去订阅
       source.subscribe(new MapObserver<T, U>(t, function));
   }
   

   接着 O1.subscribe(Obs2)，上游继续往上找，直到最初的 ObservableCreate。

2. 自上而下的数据发射： ObservableCreate 开始发数据，调用 Obs2.onNext(data)。Obs2（即 MapObserver）的 onNext 是怎么实现的呢？

   @Override
   public void onNext(T t) {
       if (done) return;
       U v;
       try {
           // 1. 应用我们传入的 mapper 函数，比如 s -> s.length()
           v = mapper.apply(t);
       } catch (Throwable ex) {
           fail(ex);
           return;
       }
       // 2. 将转换后的结果发给更下游的 Observer (Obs1)
       downstream.onNext(v);
   }
   

用图解直击本质

通过上述源码解析，我们可以画出如下的执行流程图：

sequenceDiagram
    participant App as 开发者(调用subscribe)
    participant O2 as ObservableMap
    participant O1 as ObservableCreate
    participant Obs2 as MapObserver
    participant Obs1 as 下游真正的Observer

    Note over App, O1: 1. 订阅过程（自下而上）
    App->>O2: subscribe(Obs1)
    O2->>O2: 包装Obs1生成Obs2
    O2->>O1: source.subscribe(Obs2)
    O1->>O1: 启动发射逻辑

    Note over O1, App: 2. 发射过程（自上而下）
    O1->>Obs2: onNext("Hello")
    Obs2->>Obs2: length = mapper.apply("Hello")
    Obs2->>Obs1: downstream.onNext(5)

总结： 每一次调用操作符，RxJava 都会生成一个新的 Observable 包裹住旧的 Observable。 当调用 subscribe 时，订阅动作从链条的最末端逆流而上，每一层都会生成一个新的 Observer 包裹住下游的 Observer。 当数据开始发射时，数据流从最顶端顺流而下，经过每一层 Observer 的加工，最终到达开发者定义的 Observer。

这就是 RxJava 所谓“洋葱模型”或“俄罗斯套娃”机制的真相。

总结

RxJava 并不仅仅是一个异步库，它提供了一套完整的函数式响应型编程范式。

• What：基于事件流的异步处理框架。
• How：利用高度解耦的观察者模式作为骨架，结合装饰器模式实现操作符的无限嵌套和数据加工。订阅行为自下而上层层包装，数据流动自上而下逐级处理。
• Why this way：这种设计使得复杂的异步操作可以被拆解为单一职责的“加工台”（操作符）。无论是数据转换（map）、过滤（filter）还是合并（zip），都可以通过流水线节点进行无缝拼接，彻底消灭了嵌套回调的噩梦。

在理解了 RxJava 的这套洋葱皮架构后，我们在遇到操作符组合失效、事件丢失等问题时，便能够在脑海中清晰地勾勒出数据的流向。然而，目前的流水线还只是在“同一个车间”（单线程）里运转。在实际开发中，我们往往需要将网络请求放在后台车间，将结果展示放在前台车间。

这便引出了 RxJava 另一个极其强大的特性——线程调度（Schedulers）。我们将在下一篇文章中深度剖析 RxJava 线程切换的底层奥秘。