设计模式与响应式编程：观察者模式与Flux、责任链与Operator、策略与调度器

响应式编程这几年火得不行。说实话，我最早接触它的时候，第一反应是——这不就是观察者模式套了个马甲吗？后来深入项目才发现，事情没那么简单。响应式编程不是简单套用某个设计模式，而是把多个模式揉在一起，形成了一套完整的编程范式。

今天咱们就来拆解一下，响应式编程里到底藏了哪些设计模式的影子。我会结合Java的Project Reactor和C++的RxCpp来聊，毕竟这两个库我都踩过坑。

观察者模式：响应式的基石

响应式编程最核心的东西，其实就是观察者模式。你想想看，Publisher（发布者）和Subscriber（订阅者），这不就是Subject和Observer吗？

但有个关键区别——响应式里多了背压（Backpressure）的概念。我在项目中遇到过一个问题：一个数据源每秒产生10万条数据，消费者每秒只能处理1万条。如果没有背压，内存直接爆掉。

看个Java Reactor的例子：

Flux<Integer> flux = Flux.range(1, 100)
    .onBackpressureBuffer(1000); // 缓冲区最多1000个

flux.subscribe(new BaseSubscriber<Integer>() {
    @Override
    protected void hookOnNext(Integer value) {
        // 模拟慢消费
        try { Thread.sleep(10); } catch (InterruptedException e) {}
        System.out.println("收到: " + value);
        request(1); // 手动请求下一个
    }
});

这里的关键是request(1)。消费者告诉生产者：我准备好了，再发一个过来。这就是背压的精髓——消费者控制流量，而不是生产者拼命塞。

核心要点：观察者模式在响应式编程中的进化——从"推模式"变成了"推拉结合"。消费者可以主动拉取数据，而不是被动接收。

C++的RxCpp里也有类似机制：

auto observable = rxcpp::observable<>::range(1, 100)
    .map([](int i) { return i * 2; });

observable.subscribe(
    [](int v) { std::cout << "收到: " << v << std::endl; },
    []() { std::cout << "完成" << std::endl; }
);

嗯，这里要注意，C++版本默认没有背压支持，需要自己实现。我建议你在生产环境中用的时候，一定要加上流量控制，否则很容易出问题。

责任链模式：Operator的串联艺术

响应式编程里最爽的是什么？链式调用。一个flux.map().filter().flatMap()写下来，行云流水。这背后其实就是责任链模式。

每个Operator都是一个处理器，处理完数据后传给下一个。我曾经在项目中调试过一个bug，数据经过8个Operator后结果不对。当时我一个个Operator拆开测，最后发现是flatMap里的并发问题。

看这个例子：

Flux.just("user1", "user2", "user3")
    .map(name -> name.toUpperCase())        // 第一个处理器
    .filter(name -> name.startsWith("U"))   // 第二个处理器
    .flatMap(name -> fetchUserData(name))   // 第三个处理器
    .subscribe(System.out::println);

每个Operator都像责任链里的一个节点。数据流经过时，每个节点决定：处理还是不处理？处理完传给谁？

我的经验：写Operator链的时候，尽量保持每个Operator职责单一。我曾经见过有人在一个map里干了三件事——数据转换、过滤、日志记录。这种代码调试起来特别痛苦。拆成三个Operator，每个只做一件事，出了问题一眼就能定位。

C++的RxCpp里Operator链写法类似：

auto chain = rxcpp::observable<>::range(1, 10)
    .map([](int i) { return i * 10; })
    .filter([](int i) { return i % 20 == 0; })
    .take(3);

chain.subscribe([](int v) { std::cout << v << " "; });

输出：10 30 50

策略模式：调度器的灵活切换

响应式编程里，调度器（Scheduler）决定了代码在哪个线程执行。这其实就是策略模式——不同的调度策略可以随时切换。

Java Reactor里常见的调度器：

调度器 说明 适用场景
Schedulers.immediate() 当前线程执行 简单测试
Schedulers.single() 单线程复用 定时任务
Schedulers.elastic() 弹性线程池 IO密集型
Schedulers.parallel() 固定大小线程池 CPU密集型

看个实际例子：

Flux.range(1, 100)
    .subscribeOn(Schedulers.parallel())  // 生产者在并行线程池
    .publishOn(Schedulers.elastic())     // 消费者在弹性线程池
    .subscribe(v -> {
        // 这里可能是IO操作
        saveToDatabase(v);
    });

为什么这么设计？生产者是CPU密集型的计算，用parallel。消费者是IO密集型的数据库写入，用elastic。不同的策略应对不同的场景。

我曾经踩过的坑：有一次我把所有操作都放在parallel调度器上，结果数据库连接池被打满了。因为parallel的线程数是CPU核心数，而数据库连接池需要更多线程来等待IO。后来改成elastic，问题就解决了。记住：IO密集型操作不要用parallel。

C++的RxCpp里调度器用法：

auto coordination = rxcpp::observe_on_new_thread();
auto observable = rxcpp::observable<>::range(1, 10)
    .map([](int i) { return i * 2; })
    .observe_on(coordination); // 切换到新线程

observable.subscribe([](int v) {
    std::cout << "线程ID: " 
              << std::this_thread::get_id() 
              << " 值: " << v << std::endl;
});

三种模式的协同工作

这三种模式不是孤立的。在实际项目中，它们经常一起出现。我画了一张图来说明它们的关系：

响应式编程中的设计模式协同 数据源 Publisher 观察者模式 背压控制 推拉结合 Subscriber 责任链模式 Operator链 map/filter/flatMap 职责单一 消费 策略模式：调度器 parallel() | elastic() | single() | immediate() 数据流经过观察者模式（背压控制）→ 责任链模式（Operator处理）→ 策略模式（调度器决定线程） 三种模式协同工作，缺一不可 观察者提供基础架构，责任链处理数据，策略优化性能

从这张图能看出来，数据从源头出发，经过观察者模式建立订阅关系，然后通过责任链模式的Operator链进行各种转换，最后在策略模式的调度器控制下执行。每一步都有明确的设计模式在背后支撑。

实战中的选择建议

说了这么多，到底什么时候用响应式编程？我个人的经验是：

  • 高并发IO场景：比如Web服务器、消息队列消费者。响应式编程能大幅减少线程数。
  • 数据流处理：比如实时计算、日志分析。Operator链写起来比传统回调优雅得多。
  • 异步任务编排：多个异步任务有依赖关系时，响应式比CompletableFuture更灵活。

但也不是所有场景都适合。如果你的业务逻辑是简单的CRUD，用响应式反而增加了复杂度。我见过有人为了用响应式而用响应式，结果代码变得难以维护。记住：设计模式是工具，不是目的。

最后说一句：响应式编程不是银弹。它把观察者模式、责任链模式、策略模式融合在一起，解决的是特定领域的问题。理解这些设计模式在响应式中的角色，比死记API更重要。下次你写Flux链的时候，想想每个Operator背后是什么模式在支撑——这样你才能真正用好它。

公众号:蓝海资料掘金营，微信deep3321

← 上一章 📖 返回目录 下一章 →