java

4.1 Mono 与 Flux：零个、一个与多个元素

深入讲解 Project Reactor 的两个核心类型：Mono<T>（0 或 1 个元素）与 Flux<T>（0 到 N 个元素）。覆盖创建操作符（just、fromIterable、create、defer）、empty 与 error 的语义差异、Mono<Void> 的用法，以及它们与 CompletableFuture 的真实对应关系。示例基于 Reactor BOM 2025.0.8（reactor-core 3.8.8）。

Leeting Yan 5 分钟阅读 2101 字

从手写 Publisher 到 Mono/Flux

第 1–3 章我们手写了 Publisher/Subscriber/Subscription 的实现，理解了 Reactive Streams 规范的信号契约。从本章开始，我们切换到 Project Reactor——一个基于 RS 规范的高阶抽象库。

Reactor 提供了两个核心类型：

类型元素数量类比
Mono<T>0 或 1Optional<T> 的异步版
Flux<T>0 到 NStream<T> 的异步版

这不是简单的命名差异。Mono 和 Flux 的 API 设计、操作符语义、甚至内部实现都有本质区别。理解「什么时候用 Mono」是写出正确响应式代码的第一步。

Mono：零个或一个元素

创建 Mono

import reactor.core.publisher.Mono;

// 1. 已有值
Mono<String> mono1 = Mono.just("hello");

// 2. 空 Mono（只有 onComplete）
Mono<String> mono2 = Mono.empty();

// 3. 错误 Mono（只有 onError）
Mono<String> mono3 = Mono.error(new RuntimeException("fail"));

// 4. 延迟创建（每次订阅时才调用 Supplier）
Mono<String> mono4 = Mono.defer(() -> Mono.just(fetchFromCache()));

// 5. 从 Callable（可能抛出异常）
Mono<String> mono5 = Mono.fromCallable(() -> readFile());

// 6. 从 CompletableFuture
Mono<String> mono6 = Mono.fromFuture(CompletableFuture.supplyAsync(() -> "async"));

Mono.just vs Flux.just

Mono.just(1).subscribe(System.out::println);
// 输出: 1

Flux.just(1).subscribe(System.out::println);
// 输出: 1

两者都发射 1 个元素然后完成。但 Mono.just(1) 的类型是 Mono<Integer>，保证最多 1 个元素；Flux.just(1) 的类型是 Flux<Integer>，语义上是「N 个元素中只有 1 个」。

这个区别在组合时很重要：

Mono<Integer> mono = Mono.just(1);
Flux<Integer> flux = Flux.just(1);

// Mono 可以转成 Flux（展开成 1 个元素）
Flux<Integer> fromMono = mono.flux();  // Flux.just(1)

// Flux 也可以转成 Mono（只取第一个）
Mono<Integer> fromFlux = flux.next();  // Mono.just(1)

Mono：只关心完成，不关心数据

在异步操作中，经常有一种场景：「我只想知道操作是否完成，不需要返回结果」。

// 保存记录，返回 Mono<Void>
Mono<Void> save = Mono.fromRunnable(() -> database.save(record));

// 订阅时只关心完成
save.subscribe(
    ignored -> { },                // Mono<Void> 不会调用 onNext
    error -> log.error("save failed", error),
    () -> log.info("save completed")
);

Mono<Void> 是一个特殊的语义标记：它告诉下游「这个操作要么成功完成，要么失败，没有中间数据」。这在 REST API 的 POST/DELETE 操作、消息发送确认等场景很常见。

运行验证

public class MonoDemo {
    public static void main(String[] args) {
        System.out.println("=== Mono.just ===");
        Mono.just("hello").subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e),
            () -> System.out.println("onComplete")
        );

        System.out.println("\n=== Mono.empty ===");
        Mono.empty().subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e),
            () -> System.out.println("onComplete")
        );

        System.out.println("\n=== Mono.error ===");
        Mono.error(new RuntimeException("boom")).subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e.getMessage()),
            () -> System.out.println("onComplete")
        );
    }
}

输出：

=== Mono.just ===
onNext: hello
onComplete

=== Mono.empty ===
onComplete

=== Mono.error ===
onError: boom

注意：Mono.empty() 直接调用 onComplete，不调用 onNext。Mono.error() 直接调用 onError，不调用 onNext。

Flux：零到多个元素

创建 Flux

import reactor.core.publisher.Flux;

// 1. 固定值
Flux<Integer> flux1 = Flux.just(1, 2, 3);

// 2. 从 Iterable
Flux<Integer> flux2 = Flux.fromIterable(List.of(1, 2, 3));

// 3. 从范围
Flux<Integer> flux3 = Flux.range(1, 5);  // 1, 2, 3, 4, 5

// 4. 间隔发射
Flux<Long> flux4 = Flux.interval(Duration.ofMillis(100)).take(5);

// 5. 动态生成
Flux<Integer> flux5 = Flux.create(sink -> {
    sink.next(1);
    sink.next(2);
    sink.complete();
});

// 6. 延迟生成（每次订阅时）
Flux<String> flux6 = Flux.defer(() -> Flux.just(fetchData()));

Flux.range 与背压

Flux.range(1, 100)
    .subscribe(new BaseSubscriber<Integer>() {
        @Override
        protected void hookOnSubscribe(Subscription subscription) {
            request(10);  // 先请求 10 个
        }

        @Override
        protected void hookOnNext(Integer value) {
            System.out.println("received: " + value);
            if (value % 10 == 0) {
                request(10);  // 每 10 个再请求 10 个
            }
        }
    });

Flux.range 是一个 Cold Publisher——每次订阅独立计数。它内置了背压支持：request(n) 时只发射 n 个元素。

运行验证

public class FluxDemo {
    public static void main(String[] args) {
        System.out.println("=== Flux.just ===");
        Flux.just(1, 2, 3).subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e),
            () -> System.out.println("onComplete")
        );

        System.out.println("\n=== Flux.range ===");
        Flux.range(1, 5).subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e),
            () -> System.out.println("onComplete")
        );

        System.out.println("\n=== Flux.empty ===");
        Flux.empty().subscribe(
            v -> System.out.println("onNext: " + v),
            e -> System.out.println("onError: " + e),
            () -> System.out.println("onComplete")
        );
    }
}

输出：

=== Flux.just ===
onNext: 1
onNext: 2
onNext: 3
onComplete

=== Flux.range ===
onNext: 1
onNext: 2
onNext: 3
onNext: 4
onNext: 5
onComplete

=== Flux.empty ===
onComplete

Mono 与 Flux 的转换

Mono → Flux

方法结果场景
mono.flux()1 个元素或空需要统一用 Flux API
mono.repeat()重复 N 次轮询重试
mono.expand(...)递归展开链表遍历

Flux → Mono

方法结果场景
flux.next()第一个元素只取一个
flux.collectList()列表聚合全部
flux.count()元素数量统计
flux.reduce(...)归约值求和/求积
// Flux 转 Mono（取第一个）
Mono<Integer> first = Flux.range(1, 10).next();

// Flux 转 Mono（聚合）
Mono<List<Integer>> list = Flux.range(1, 5).collectList();
Mono<Integer> sum = Flux.range(1, 5).reduce(0, Integer::sum);

什么时候选 Mono，什么时候选 Flux

场景选择原因
HTTP GET 单个资源Mono<Resource>0 或 1 个结果
HTTP GET 列表Flux<Resource>0 到 N 个结果
HTTP POST/PUT/DELETEMono<Void>只关心成功/失败
数据库查询（单行）Mono<Row>0 或 1 个结果
数据库查询（多行）Flux<Row>0 到 N 个结果
消息发送Mono<Void>发送完成确认
消息消费Flux<Message>持续消费
定时任务Mono<String>执行一次返回结果
事件流Flux<Event>持续事件

一个常见的错误：

// 错误：用 Flux 表示确定只有一个元素
public Flux<User> findById(String id);  //  confusing！应该用 Mono<User>

// 正确
public Mono<User> findById(String id);
public Flux<User> findAll();

底层：Mono 和 Flux 都是 Publisher

虽然 Mono 和 Flux 提供了丰富的 API，但它们的底层都实现了 org.reactivestreams.Publisher：

// Mono 实现了 Publisher
Publisher<String> publisher = Mono.just("hello");

// Flux 也实现了 Publisher
Publisher<Integer> publisher2 = Flux.range(1, 5);

这意味着你可以把 Mono/Flux 传给任何接受 Publisher 的 API，包括我们第 2–3 章手写的 LoggingSubscriber。

Mono.just("hello").subscribe(new LoggingSubscriber<>("test"));
// 输出: [main][test] onNext: hello
// 输出: [main][test] onComplete

与 CompletableFuture 的对照

CompletableFutureMono差异
CompletableFuture<T>Mono<T>Future 的取消不会向上游传播；Mono 的 cancel 会
CompletableFuture<Void>Mono<Void>只关心完成/失败
CompletableFuture<List<T>>Flux<T> + collectList()前者已聚合，后者是逐个元素
thenApplymap同步转换
thenComposeflatMap异步展开
exceptionally(f)onErrorResume(e -> Mono.just(f.apply(e)))exceptionally 的 f 返回值，onErrorResume 要求返回 Publisher；固定值用 onErrorReturn
timeout（JDK 9+ 的 orTimeout）timeout语义相近，Reactor 还可切换备用流
cancelcancelCompletableFuture 也有 cancel，但不会停止上游任务

从 CompletableFuture 迁移到 Mono 通常是平滑的，特别是当你已经有了异步操作的基础。

小结

类型元素数创建方式终止信号
Mono<T>0 或 1just、empty、error、deferonNext + onComplete 或 onError
Flux<T>0 到 Njust、range、interval、createonNext × N + onComplete 或 onError

选择 Mono 还是 Flux 取决于你的数据语义：

  • Mono： HTTP 单资源查询、数据库单行、保存操作确认、定时任务结果。
  • Flux： HTTP 列表查询、数据库多行、消息流、事件流、定时心跳。

下一节，我们进入 Reactor 最核心的操作符：map、flatMap 和 concatMap——它们决定了数据如何在流中转换和展开。

阅读导航

上一篇：3.3 调度切换：subscribeOn 与 publishOn

下一篇：4.2 转换操作符：map、flatMap、concatMap

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章 返回首页

「java」更多文章

  1. 15.3 全书总结与学习路径回顾
  2. 15.2 响应式流处理的演进趋势
  3. 15.1 引擎选型指南