在使用 Project Reactor 处理响应式数据流时，doOnError 本身无法访问上游已发出但尚未被消费的元素；若需在错误发生时记录具体失败项，应改用 flatMap 配合 doOnError（针对保存操作）或 onErrorResume，确保错误上下文与当前数据项绑定。
在使用 project reactor 处理响应式数据流时，`doonerror` 本身无法访问上游已发出但尚未被消费的元素；若需在错误发生时记录具体失败项，应改用 `flatmap` 配合 `doonerror`（针对保存操作）或 `onerrorresume`，确保错误上下文与当前数据项绑定。
在响应式编程中，一个常见误区是认为 doOnError 能“回溯”获取导致错误的数据项。实际上，doOnError 只接收 Throwable，它作用于整个流的终止错误，不持有触发该错误的上游元素——尤其当错误发生在 doOnNext 后续的异步/阻塞操作（如数据库保存）中时，原始 Item 已脱离当前操作符的作用域。
要实现“记录哪个 Item 保存失败”，关键在于将错误处理逻辑下沉到每个元素的独立处理链中。以下是两种典型场景及对应方案：
✅ 场景一：saveRecord 是响应式操作（返回 Mono）
此时应使用 flatMap 展开每个 Item，并在其专属的 Mono 上附加错误处理：
reactiveTemplate
.query(cql, itemMapper) // Flux<item>
.flatMap(item ->
saveRecord(item) // Mono<void>
.doOnError(error -> {
log.error("Failed to save item: {}, error: {}", item, error.getMessage());
processErrorRecord(item, error); // 传入 item + 异常
})
.onErrorResume(error -> {
// 可选：兜底处理，如返回空 Mono 或降级结果
return Mono.empty();
})
);</void></item>
? 提示：doOnError 用于副作用（如日志、监控），而 onErrorResume 用于错误恢复。若 processErrorRecord 本身是响应式（返回 Mono
），应直接链式调用 .onErrorResume(e -> processErrorRecord(item, e))。
✅ 场景二：saveRecord 是阻塞式同步方法（返回 void 或非响应式类型）
必须将其包装为 Mono 并指定线程调度器，避免阻塞事件循环：
reactiveTemplate
.query(cql, itemMapper) // Flux<item>
.flatMap(item ->
Mono.fromRunnable(() -> saveRecord(item))
.subscribeOn(Schedulers.boundedElastic()) // ⚠️ 必须！避免阻塞 IO 线程
.doOnError(error -> {
log.error("Blocked save failed for item: {}, cause: {}", item, error.toString());
processErrorRecord(item, error);
})
.then() // 转换为 Mono<void>
);</void></item>
⚠️ 注意：切勿省略 subscribeOn(Schedulers.boundedElastic())。boundedElastic() 专为阻塞 I/O 设计，能自动扩容并避免线程耗尽。
❌ 错误示范（为什么 doOnNext + doOnError 不行？）
// ❌ 错误：error 发生在 saveRecord 内部，doOnError 捕获的是流级异常，item 已不可达
flux.doOnNext(this::saveRecord)
.doOnError(e -> processErrorRecord(???, e)); // ← 这里无法拿到 item
总结建议
-
永远优先让业务逻辑响应式化：将 saveRecord 改为返回 Mono
，这是最符合 Reactor 设计哲学的方式； - 阻塞调用必须隔离线程：使用 boundedElastic() 调度器，并确保 processErrorRecord 不引入新阻塞；
- 错误处理粒度要匹配数据粒度：对每个 Item 单独建模错误路径，而非依赖全局流错误钩子；
- 日志务必包含上下文：在 doOnError 中显式打印 item.toString() 或关键字段（如 item.getId()），便于问题定位。
通过将错误处理内联到每个元素的处理链中，你不仅能精准捕获失败项，还能灵活控制重试、降级或补偿逻辑，真正发挥响应式编程的可观测性与弹性优势。
