本文介绍如何利用 Project Reactor 的 Mono.expand() 操作符，优雅、非阻塞地将依赖前序结果的单值异步计算（Mono）递归展开为有序数据流（Flux），避免 block() 阻塞调用和深层 flatMap 嵌套风险。
本文介绍如何利用 project reactor 的 `mono.expand()` 操作符，优雅、非阻塞地将依赖前序结果的单值异步计算（mono）递归展开为有序数据流（flux），避免 `block()` 阻塞调用和深层 `flatmap` 嵌套风险。
在响应式编程中，当后续数据库查询严格依赖前一次查询结果（例如分页游标、递归树遍历、状态机推进等场景）时，开发者常面临一个核心挑战：如何将「产生下一个 Mono」的函数安全、高效地转化为持续发射元素的 Flux
此时，Mono.expand() 是专为此类递归式、结果驱动型流生成设计的理想方案。它接受一个 Function
以下是一个典型实现：
private Mono<Integer> calculateNext(Integer value) {
return Mono.defer(() -> Mono.just(value + 1))
.delayElement(Duration.ofSeconds(1L)); // 模拟异步 DB 查询延迟
}
private Flux<Integer> generateSequence(int start, int endInclusive) {
return calculateNext(start)
.expand(next -> next < endInclusive
? calculateNext(next)
: Mono.empty()) // 终止条件：返回 empty 结束展开
.takeUntil(i -> i >= endInclusive); // 双重保障终止（也可仅用此行）
}调用示例：
generateSequence(0, 5)
.doOnNext(i -> LOG.info("Emitting: {}", i))
.subscribeOn(Schedulers.boundedElastic()) // 若 calculateNext 含阻塞 I/O，需指定线程池
.blockLast(); // 仅测试用；生产环境应链式处理（如 map/flatMap/collectList）输出日志：
Emitting: 1 Emitting: 2 Emitting: 3 Emitting: 4 Emitting: 5
✅ 关键优势说明：
- 零阻塞：expand() 内部基于 flatMap + concatMap 优化实现，不调用任何 block()；
- 内存友好：展开是惰性、逐个进行的，不会预加载全部中间 Mono；
- 天然终止控制：通过返回 Mono.empty() 或配合 takeUntil / takeWhile 精确控制流长度；
- 错误传播明确：任一 calculateNext 发生错误，整个 Flux 将立即以 onError 终止；
-
可组合性强：返回的 Flux
可无缝接入任意 Reactor 操作链（如 bufferTimeout, retryWhen, onErrorResume）。
⚠️ 注意事项：
- 切勿在 expand 的 mapper 函数中引入副作用（如修改外部状态），因其执行时机受下游订阅与背压影响；
- 若 calculateNext 可能返回 null，需提前用 filter(Objects::nonNull) 或 handle() 处理，避免 NullPointerException；
- 对于超长序列（如 >10⁵ 项），建议添加超时控制：.timeout(Duration.ofMinutes(5))；
- expand() 不适用于需要“并行展开多个分支”的场景（此时应考虑 flatMap + Flux.generate 组合）。
总结：Mono.expand() 是 Project Reactor 中处理「结果驱动递归异步流」的声明式首选方案。它以简洁 API 封装了复杂的递归调度逻辑，在保证响应式语义完整性的同时，显著提升代码可读性与运行时健壮性。
