本文介绍如何基于 reactor 的 mono 流式组合能力，将多个有序的 `seedpreprocessor` 串联成一个响应式处理流水线，实现输入文档的逐级异步转换。
在响应式编程中，构建可组合、可扩展的处理链（Pipeline）是常见需求。Spring WebFlux 提供了强大的 Mono 和 Flux 操作符，其中 flatMap 是实现“上一阶段输出作为下一阶段输入”的关键——它支持异步、非阻塞地将前序结果映射为新的 Mono，并自动扁平化嵌套流。
回到你的 PipeLine 实现，核心问题在于：原代码中使用 Flux.fromIterable(...).map(...) 试图一次性对所有处理器应用 process()，但 map 是同步、并发无关的转换操作，无法传递中间状态；更重要的是，proc.process(initial) 始终以原始 initial 为输入，而非前一个处理器的输出，导致无法形成真正的链式调用。
✅ 正确做法是累积式构建 Mono 链：从初始 Mono
出发，依次 flatMap 到每个处理器，使每个 process(I) 的输入动态变为前一步的输出：</p><div class="aritcle_card flexRow">
<div class="artcardd flexRow">
<a class="aritcle_card_img" href="/xiazai/skill10835" title="React 组件生成器"><img
src="https://img.php.cn/upload/skill/001/246/273/6adaf20dee630206.jpg" alt="React 组件生成器" ></a>
<div class="aritcle_card_info flexColumn">
<a href="/xiazai/skill10835" title="React 组件生成器">React 组件生成器</a>
<p>根据项目规范生成 React 和 Vue 组件，包括 TypeScript 类型定义、Ant Design 5 集成、Tailwind CSS 样式及 Vitest 测试文件。当用户提出创建、生成或搭建前端组件的需求时启用，例如：“帮我写一个表格组件”、“生成一个弹窗”、“create a form component”、“写一个列表页”、“生成一个下拉选择器”。该能力在以下任一场景触发：涉及 React/TSX/Vue 组件生成、UI 组件创建，或为组件生成测试文件。</p>
</div>
<a href="/xiazai/skill10835" title="React 组件生成器" class="aritcle_card_btn flexRow flexcenter"><b></b><span>下载</span> </a>
</div>
</div><pre class="brush:php;toolbar:false;">public Mono<pre class="brush:php;toolbar:false;"ProcessorDocument> execute(String url) {
log.info("Start processing URL = {}", url);
PreProcessorDocument initial = new PreProcessorDocument(url);
Mono<pre class="brush:php;toolbar:false;"ProcessorDocument> result = Mono.just(initial);
for (SeedPreProcessor<pre class="brush:php;toolbar:false;"ProcessorDocument> processor : allProcessors) {
result = result.flatMap(processor::process);
}
return result;
}
? 这种写法天然具备以下优势：
- 完全响应式：全程无阻塞，支持背压，适配高并发 WebFlux 场景；
- 顺序可控：依赖构造时已按 order() 排序的 allProcessors 列表，确保执行顺序严格一致；
- 错误传播自然：任一 Processor 返回 Mono.error() 或抛出异常，都会中断流水线并向下透传；
- 资源轻量：不创建中间集合或额外线程，仅通过函数式组合构建声明式流。
⚠️ 注意事项：
- 确保所有 SeedPreProcessor 实现类的 process() 方法真正返回异步完成的 Mono（例如调用 WebClient, ReactiveMongoTemplate 等），避免意外阻塞（如误用 .block()）；
- 若需在某处理器失败后降级处理（如默认值兜底），可在对应 flatMap 后追加 .onErrorResume(...)；
- 如需记录各阶段耗时或日志，推荐使用 doOnNext, doOnSubscribe, elapsed() 等副作用操作符，而非侵入业务逻辑。
总结：响应式管道的本质不是“遍历处理器”，而是“构建数据流拓扑”。通过 flatMap 的链式累积，你获得的不仅是一段可运行代码，更是一个声明式、可观测、可组合的响应式处理单元——这正是 WebFlux 流水线设计的精髓所在。
