1.1 异步编程的演进

同步阻塞：最简单也最昂贵

在 Java 的早期年代，处理 I/O 的标准方式是同步阻塞：

Socket socket = new Socket("api.example.com", 80);
InputStream in = socket.getInputStream();
int b;
while ((b = in.read()) != -1) {   // 阻塞！线程挂起直到数据到达
    process(b);
}

每来一个请求就开一个线程。在连接数少的时候没问题——代码直来直去，容易理解。但当并发上升到数千时，操作系统的线程调度开销会让 CPU 大量时间花在上下文切换上，而不是处理业务逻辑。

Java 1.5 引入了 java.util.concurrent，ExecutorService 和 Future 让线程复用成为可能：

ExecutorService executor = Executors.newFixedThreadPool(10);
Future<String> future = executor.submit(() -> fetchFromNetwork());
String result = future.get();  // 仍然阻塞！主线程在这里等待

Future.get() 把异步变成了「在另一个地方阻塞」。你拿到了一个代表「将来会有结果」的句柄，但获取结果的那一刻，当前线程还是要停下来等。

回调函数：把阻塞拍扁

JavaScript 社区首先大规模使用回调模式解决阻塞问题。核心思想是：「结果准备好了你叫我，我先干别的。」

fetchUser(userId, user -> {
    fetchOrders(user.getAccountId(), orders -> {
        validateInventory(orders, valid -> {
            chargePayment(valid.getTotal(), receipt -> {
                log.info("receipt: {}", receipt);
            });
        });
    });
});

这就是著名的「回调地狱（Callback Hell）」。代码不是从上往下读，而是从外往里缩进。错误处理更麻烦——每个回调层都要写异常分支，而且异常往上抛的路径被回调层隔断了。

更严重的是：回调函数丢失了控制流。 你无法在两步之间插入「如果第二步超过 3 秒就放弃」这样的逻辑，因为第二步的代码在另一个回调闭包里。

Future/Promise：组合与仍然缺失的背压

Java 8 的 CompletableFuture 把回调拍扁成了链式调用：

fetchUser(userId)
    .thenCompose(u -> fetchOrders(u.getAccountId()))
    .thenCompose(this::validateInventory)
    .thenCompose(v -> chargePayment(v.getTotal()))
    .thenAccept(r -> log.info("done: {}", r))
    .exceptionally(ex -> {
        log.error("failed", ex);
        return null;
    });

链式调用的好处是：

1. 错误集中处理：exceptionally 可以捕获链上任意一步的错误。
2. 组合语义明确：thenCompose（flatMap）和 thenApply（map）的语义和 Stream 对应。
3. 不占用调用线程：发起者线程在提交 fetchUser 后就可以去干别的。

但 CompletableFuture 仍然是一种推模型（Push Model）：

// 上游一旦产生结果，立刻向下游推送
CompletableFuture<Integer> upstream = CompletableFuture.supplyAsync(
    () -> computeIntensive(), executor);

upstream.thenAccept(result -> {
    // 如果这一步处理很慢，结果已经在内存里等待了
    slowProcess(result);
});

问题在于：下游 slowProcess 一秒钟只能处理 10 个结果，而上游 computeIntensive 一秒钟能产生 1000 个。中间那 990 个结果去哪了？

答案是：它们被隐式地缓冲在 CompletableFuture 的内部状态里。CompletableFuture 的每个阶段都会持有对下游的引用，当链够长时，这些中间对象会把堆填满。

这不是 CompletableFuture 特有的问题。所有推模型共享同一个根本矛盾：生产者不知道消费者有多快。

Stream API：惰性求值，但仍是同步的

Java 8 同时引入了 Stream API，它提供了一种和集合不同的数据处理方式：

List<Integer> numbers = List.of(1, 2, 3, 4, 5);
numbers.stream()
       .filter(n -> n % 2 == 0)
       .map(n -> n * n)
       .forEach(System.out::println);

Stream 的特点是惰性求值（Lazy Evaluation）：filter 和 map 不会在调用时立刻执行，而是构建一个「操作管道」。等到终端操作（如 forEach、collect）触发时，数据才从源头逐个流过管道。

这和响应式流有相似之处，但 Stream 有三个关键限制：

1. 同步执行：forEach 在当前线程同步消费每个元素，不能异步拉取。
2. 一次性：Stream 只能消费一次，不能重复订阅。
3. 无限流困难：虽然有 Stream.iterate 和 Stream.generate，但没有内置的「消费者说停就停」的机制。

// Stream 的无限流：Consumer 无法说「停」
Stream.generate(() -> random.nextInt())
      .forEach(System.out::println);  // 永不停歇

想在 Stream 里实现背压？你需要自己用一个 BlockingQueue 做缓冲，再写一套线程协调逻辑。这本质上是在重复发明 Reactive Streams 规范。

推模型 vs 拉模型 vs 混合模型

为了理解 Reactive Streams 的设计动机，我们先对比三种数据流模型：

混合模型的核心创新是：消费者通过 request(n) 信号告诉生产者「我可以处理 n 个」，生产者只在收到请求后才发射 n 个元素。 这样，消费者既不需要阻塞等待（像拉模型），也不会被生产者淹没（像推模型）。

为什么 CompletableFuture 不够：一个实验

我们用代码证明推模型的缓冲问题。

import java.util.concurrent.*;

public class PushBufferDemo {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        for (int i = 0; i < 1_000_000; i++) {
            final int value = i;
            CompletableFuture.supplyAsync(() -> value, executor)
                .thenAccept(v -> {
                    // 模拟慢消费者：每处理一个睡 1 毫秒
                    try { Thread.sleep(1); } catch (InterruptedException e) {}
                });
        }

        System.out.println("Submitted all tasks");
        executor.shutdown();
    }
}

这个程序在 1 秒内提交了 100 万个 CompletableFuture，但消费者每毫秒才能处理一个。结果是：

1. 内存迅速攀升——每个未完成的 CompletableFuture 都持有值引用。
2. 最终可能 OOM，取决于 -Xmx 大小。
3. 线程池队列也会积压（SingleThreadExecutor 用无界 LinkedBlockingQueue）。

这不是代码写得不好。这是推模型在没有速率协商时的必然结果。

从 Observable 到 Reactive Streams

2012–2014 年间，Netflix 开源了 RxJava（受微软 Reactive Extensions 启发），首次在 JVM 上提供了基于 Observable 的响应式流：

Observable.interval(1, TimeUnit.MILLISECONDS)
    .subscribe(i -> {
        // 每毫秒来一个
        slowProcess(i);
    });

RxJava 引入了 Subscriber 和 Subscription 的概念，允许消费者通过 request(n) 控制速率。但 RxJava 1.x 的 Subscriber 接口与后来成为标准的 Reactive Streams 略有不同——它更偏向「事件处理」而非「信号契约」。

2015 年，Netflix、Pivotal、Lightbend、Red Hat 联合发布了 Reactive Streams 规范。它不是框架，不是库，而是一组跨框架的接口契约。任何实现这些接口的库（Reactor、RxJava、Akka Streams）都可以互相连接。

小结：每一层解决了什么，留下了什么

本书的目标不是推翻你已经掌握的异步工具，而是让你在使用时知道边界在哪里。如果你的系统只是偶尔发一条 HTTP 请求等一个数据库结果，CompletableFuture 仍然是最简单正确的选择。但如果你面对的是持续演化的数据流、上下游速率不匹配、需要显式控制内存占用的场景，Reactive Streams 的背压契约就是不可替代的。

下一节，我们从 Reactive Streams 规范的四个接口与 43 条规则开始，逐条建立这份契约的精确语义。

阅读导航

上一篇：前言：环境、项目与读法

下一篇：1.2 Reactive Streams 规范：四个接口

与其他异步模型的对比

历史上异步模型的演进

1995: Java 1.0 只有同步 I/O
2002: Java 1.4 NIO 引入非阻塞 I/O
2004: Java 5 Concurrent 包引入 Future
2014: Java 8 Stream + CompletableFuture
2015: Reactive Streams 1.0 发布
2017: JDK 9 Flow API
2018: Reactor 3.1 成熟
2020: RxJava 3.0 发布

每一次演进都是为了解决「如何更好地控制异步数据流」的问题。

本机实测数据

以下数据来自本机 Temurin 25.0.5+8，仅供参考：

注意：这些数字高度依赖于具体场景，不应作为性能基准。