14、观察者模式进阶：响应式编程中的观察者模式、RxJava与观察者模式

好，咱们今天聊点硬核的。

前面几章我们把观察者模式的基本套路讲透了。但说实话，传统观察者模式在实际项目中，用起来总有点「不够爽」的感觉。你想想看，一个按钮点击触发一堆回调，数据流要手动管理订阅和取消订阅，多线程环境下还得自己处理同步……嗯，我早年在一个金融项目中就吃过这个亏。

传统观察者模式的痛点

先说说我踩过的坑。那时候做一个实时行情系统，用传统观察者模式监听股票价格变化。代码写出来是这样的：

// 传统观察者模式
public class StockPriceSubject {
    private List<Observer> observers = new ArrayList<>();
    private double price;
    
    public void attach(Observer o) { observers.add(o); }
    public void detach(Observer o) { observers.remove(o); }
    
    public void setPrice(double price) {
        this.price = price;
        notifyObservers();
    }
    
    private void notifyObservers() {
        for (Observer o : observers) {
            o.update(price);
        }
    }
}

看着挺规整对吧？但实际跑起来问题就来了：

• 数据流不可控：价格更新频率一高，观察者根本来不及处理
• 线程问题：价格推送来自网络线程，UI更新必须在主线程，手动切换太痛苦
• 错误处理缺失：某个观察者抛异常，整个通知链就断了
• 背压问题：生产者太快，消费者太慢，内存直接爆掉

我当时就在想，有没有一种模式，能把观察者模式「升级」一下，让它能处理这些现实问题？

响应式编程：观察者模式的「进化版」

说白了，响应式编程就是观察者模式 + 迭代器模式 + 函数式编程的合体。它把数据看作一条「流」，你可以对这个流做各种操作——过滤、转换、合并、分流。

核心思想就四个字：一切皆流。

我画了一张图，帮你理解这个进化关系：

RxJava 中的观察者模式

RxJava 是响应式编程在 Java 世界最经典的一个实现。它的核心接口，说白了就是观察者模式的「豪华升级版」。

咱们先看最基础的几个角色：

传统观察者	RxJava 对应	说明
Subject	Observable	数据源，发射数据流
Observer	Observer / Subscriber	订阅者，接收数据
attach()	subscribe()	建立订阅关系
update()	onNext() / onError() / onComplete()	三个回调方法

看到没？RxJava 把观察者模式里的 update() 拆成了三个方法。为什么要这么干？

我当年做项目时，最头疼的就是「数据流什么时候结束」这个问题。传统观察者模式里，观察者永远不知道数据流是否终止。RxJava 的 onComplete() 就解决了这个问题。还有 onError()，让错误处理变得优雅多了。

实战：用 RxJava 重构股票行情系统

还记得我前面说的那个行情系统吗？用 RxJava 重写之后，代码清爽多了：

// RxJava 版本的股票价格监听
Observable<Double> priceStream = Observable.create(emitter -> {
    // 模拟价格推送
    new Thread(() -> {
        for (int i = 0; i < 100; i++) {
            double price = 100 + Math.random() * 10;
            emitter.onNext(price);
            Thread.sleep(100);
        }
        emitter.onComplete();
    }).start();
});

// 订阅并处理
priceStream
    .filter(price -> price > 105)  // 只关注价格超过105的
    .map(price -> String.format("当前价格: %.2f", price))
    .subscribe(
        msg -> System.out.println(msg),  // onNext
        err -> System.err.println("出错: " + err),  // onError
        () -> System.out.println("数据流结束")  // onComplete
    );

操作符：观察者模式的「超能力」

操作符是 RxJava 最强大的地方。它本质上是对观察者模式的一种「装饰」——每个操作符都包装了原始的 Observable，返回一个新的 Observable。

我常用的几个操作符：

操作符	作用	类比传统观察者
map()	转换数据	手动在 update() 里做转换
filter()	过滤数据	在 update() 里加 if 判断
merge()	合并多个数据流	需要自己维护多个 Subject
debounce()	防抖，控制频率	传统模式很难优雅实现
subscribeOn()	指定数据生产线程	传统模式需要手动 new Thread()
observeOn()	指定数据消费线程	传统模式需要 Handler/runOnUiThread

背压：观察者模式没解决的问题

传统观察者模式有一个致命缺陷：如果生产者比消费者快，数据会堆积。我当年在行情系统里就遇到过，价格更新太快，UI 线程根本来不及刷新，最后内存溢出。

RxJava 用背压（Backpressure）机制解决了这个问题。说白了就是：消费者告诉生产者「你慢点，我吃不消了」。

// 带背压的 Observable
Flowable<Double> priceFlow = Flowable.create(emitter -> {
    for (int i = 0; i < 1000; i++) {
        emitter.onNext(Math.random() * 10);
    }
    emitter.onComplete();
}, BackpressureStrategy.BUFFER);  // 背压策略

priceFlow
    .onBackpressureDrop()  // 消费不过来就丢弃
    .observeOn(Schedulers.computation())
    .subscribe(System.out::println);

Subject：既是观察者，又是被观察者

RxJava 里有一个特殊的角色叫 Subject。它既可以作为 Observable 发射数据，又可以作为 Observer 订阅其他数据源。

我常用的是 PublishSubject：

// PublishSubject 示例
PublishSubject<String> subject = PublishSubject.create();

// 订阅者1
subject.subscribe(msg -> System.out.println("订阅者1: " + msg));
// 订阅者2
subject.subscribe(msg -> System.out.println("订阅者2: " + msg));

// 发射数据
subject.onNext("Hello");
subject.onNext("World");

这个模式在事件总线（Event Bus）里用得特别多。我做过一个模块通信系统，就是用 Subject 做中介，各个模块之间完全解耦。

从 RxJava 到现代响应式框架

RxJava 的思想影响了很多现代框架。比如：

• Spring WebFlux：基于 Reactor（响应式流规范），底层就是观察者模式
• Android 的 LiveData：Google 官方推荐的观察者模式实现
• Java 9 的 Flow API：标准库级别的响应式流支持

但不管框架怎么变，核心思想没变：数据流 + 操作符 + 订阅者。你只要理解了观察者模式，理解这些框架就只是语法层面的差异了。