Flux.doOnEach与doOnNext结合publishOn的行为差异探究

核心功能区别 #

• doOnNext：仅在流**推送元素（onNext信号）**时执行回调逻辑，元素被过滤、丢弃后，下游的doOnNext不会收到该元素的信号。
• doOnEach：对流中所有信号类型都会响应，包括元素推送（onNext）、流完成（onComplete）、流错误（onError）。哪怕元素被过滤，只要流最终发送onComplete，doOnEach也会捕获这个信号。

结合测试场景的具体表现 #

你的测试链路是：Flux.just(元素) → doOnEach → publishOn → 两个doOnEach → filter(非空)

• 当使用doOnNext时：上游doOnNext会捕获所有原始元素的onNext信号，下游两个doOnNext因元素被过滤不会触发，结果与你预期的元素数量对应触发次数一致。
• 当使用doOnEach时：
  1. 上游doOnEach会捕获每个原始元素的onNext信号，但下游两个doOnEach因元素被过滤，无法收到这些onNext信号。
  2. 流最终发送onComplete信号时，三个doOnEach都会触发回调，但如果测试未正确等待publishOn线程池的异步操作完成，下游两个doOnEach的onComplete触发会被遗漏，最终只统计到1次触发，与预期不符。

关键总结 #

如果仅需针对实际推送的元素执行回调，优先用doOnNext；如果需要处理流的生命周期信号（完成、错误），再使用doOnEach。在异步场景（如publishOn）下，测试必须确保等待所有异步操作完成，避免信号遗漏。

内容的提问来源于stack exchange，提问作者Adam