Stream API 与 Flux：Java 开发者的差异 | 指南

Stream API 与 Flux：架构差异与实际应用

理解 Stream API 和 Flux 的差异，对于处理异步系统的 Java 开发者来说至关重要。尽管方法（如 filter、map、limit）相似，但这些工具解决的是根本不同的问题。Stream API 设计用于同步处理有限集合，而 Project Reactor 中的 Flux 则专为响应式流数据处理而构建，支持背压和异步。本文剖析了关键架构差异，并帮助你选择合适的工具。

Stream API 基础：同步集合处理

Stream API 是 Java 内置的声明式集合处理机制。其核心特性：在单个执行线程中处理有限数据集。调用 .stream() 会创建一个顺序流，在发起线程中处理元素（例如主线程）。

重要的是要记住操作分类：

• 中间操作（filter、map、flatMap、limit、skip）——构建转换链，而不立即执行。它们返回新流，可以无限链式调用。
• 终端操作（collect、forEach、reduce）——触发整个链并关闭流。调用后，流无法重用。

考虑一个过滤和转换数据的示例：

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

这里：

• filter 保留偶数
• map 将每个元素加倍
• forEach 触发执行并打印结果

Stream API 的一个关键限制是单次使用。尝试另一个终端操作会抛出 IllegalStateException。这是与响应式流的一个根本区别。

Flux：响应式流数据处理

Project Reactor 中的 Flux 实现了 Reactive Streams 规范，用于异步处理无限数据流。与 Stream API 不同，Flux 支持：

• 非阻塞执行
• 背压（数据消费速率控制）
• 通过响应式操作符处理错误
• 任意线程中的事件订阅

关键架构差异：

• 终端操作——必须调用 .subscribe() 来启动处理。没有它，Flux 保持“冷”状态，不会生成数据。
• 惰性初始化——仅在订阅后生成数据。
• 无限流支持——Flux 处理无限源（例如实时传感器数据）。

Flux 中的等效操作：

Flux.just(1, 2, 3, 4, 5)
    .filter(n -> n % 2 == 0)
    .map(n -> n * 2)
    .subscribe(System.out::println); // 4, 8

这里，.subscribe() 不仅启动处理，还管理流生命周期。Flux 使用 .onErrorResume() 处理错误，并用 .limitRate() 节流。

关键差异：比较表格

为清晰起见，这里是关键差异的结构化分解：

• 执行模型

- Stream API：同步、阻塞处理

- Flux：异步、非阻塞处理，支持背压

• 数据类型

- Stream API：仅有限集合

- Flux：有限和无限流

• 错误处理

- Stream API：直接抛出异常（需要 try/catch）

- Flux：内置操作符错误处理（onErrorResume、retry）

• 可重用性

- Stream API：终端操作后单次使用

- Flux：正确设置后可多次订阅同一流

• 并行性

- Stream API：.parallelStream() 跨 CPU 核心分发

- Flux：.publishOn() 和 .subscribeOn() 控制执行上下文

实际用例：真实场景中的数据转换

考虑更新员工及其级别集合（使用 HashMap）。需求：

• 对于初级员工，将级别升级为中级
• 添加“(已晋升)”标签
• 返回更新后的结构

Stream API 实现：

Map<String, String> updatedEmployees = employees.entrySet().stream()
    .map(entry -> {
        String grade = entry.getValue();
        if ("Junior".equals(grade)) {
            return Map.entry(entry.getKey(), "Middle (increased)");
        }
        return entry;
    })
    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));

使用 Flux 的相同场景：

Map<String, String> updatedEmployees = Flux.fromStream(employees.entrySet().stream())
    .map(entry -> {
        if ("Junior".equals(entry.getValue())) {
            return Map.entry(entry.getKey(), "Middle (increased)");
        }
        return entry;
    })
    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
    .block();

关键实现差异：

• Flux 需要显式 .block() 来获取结果（违背响应式原则）
• Stream API 阻塞执行线程直到完成
• 对于响应式系统，优先使用 .subscribe() 实现非阻塞

关键要点

• Stream API——适合单线程同步有限集合处理。用于 ETL 操作或业务逻辑中的数据转换。
• Flux——处理异步流数据。完美适用于网络请求、传感器或事件总线。
• 背压——Flux 的关键特性，Stream API 缺少此功能。可在高负载下保持系统稳定。
• 错误处理——Flux 中响应式，Stream API 中传统 try/catch。
• 性能——短同步操作用 Stream API 胜出；长异步进程用 Flux。

— Editorial Team