这是一种响应式流生命周期误用导致的信号停滞现象,表现为上游已发数据但下游收不到onnext、oncomplete或onerror;根本原因是controller.enqueue未被调用,致背压未履行、数据无法流动。

这不是数据流死锁(Deadlock),也不是 JVM 层面的线程死锁,而是一种**响应式流生命周期误用导致的信号停滞现象**——具体表现为:上游已产生数据,下游 Subscriber 却收不到 onNext,也没有 onComplete 或 onError,整个链路“静默卡住”。根本原因不是锁竞争,而是 controller.enqueue 未被调用,导致自定义 Subscriber 或 Subscription 实现中背压未被履行、数据无法向下流动。
确认是否真为 controller.enqueue 缺失问题
先排除其他常见干扰:
- 检查是否误用了
block()或未设超时的toFuture().get(),造成主线程阻塞而非流停滞 - 确认不是
filter全不匹配、switchMap切换过快、或take(1)等短路操作符提前终止 - 用
log()或doOnEach()观察信号流:若上游onNext已发出,但下游onNext完全未触发,且无onComplete/onError,才需聚焦controller
controller.enqueue 未调用的典型场景
该问题多出现在手动实现 Processor、自定义 Subscription,或使用低阶 API(如 create、push)时:
- 在
onSubscribe(Subscription s)中保存了s,但后续未调用s.request(n),导致上游因背压为 0 而暂停发送 - 在
request(long n)实现中漏掉对controller.enqueue()的调用,或只在特定条件分支中调用 - 使用
Flux.create(emitter -> { ... })时,在异步回调(如 HTTP 响应处理)里忘了调用emitter.next(data),等同于没调用enqueue - 在
onNext处理中抛出未捕获异常,且未调用controller.error(e)或controller.complete(),导致流状态悬置
快速验证与定位方法
不依赖日志猜测,直接观测控制流行为:
- 在
onSubscribe和request方法内加断点或log::info,确认request(n)是否被触发、参数是否为正数 - 在所有可能路径上(包括异常分支)插入
if (controller != null && !controller.isCancelled()) controller.next(data),避免空指针或取消后误发 - 用
materialize()包裹流,打印每个Signal类型,重点看是否有ON_NEXT信号丢失,或ON_COMPLETE出现在预期之前 - 若使用 Project Reactor,启用调试模式:
-Dreactor.trace.cancel=true,可追踪 cancel 传播是否意外中断了请求链
修复关键原则
核心是守住 Reactive Streams 规范的契约:
-
request 必须驱动 enqueue:每次
request(n)都应对应最多 n 次controller.next(),不可跳过、不可累积、不可乱序 -
cancel 后禁止再调用 controller 任何方法:务必在调用前检查
controller.isCancelled() -
异常必须显式通知:不能仅打印日志,必须调用
controller.error(e),否则流会挂起且无提示 - 优先使用高阶操作符(如
flatMap、concatMap、handle)替代手写create,减少底层控制出错概率










