Reactor 核心实验（U26）

独立新增工程：Liberica JDK21.0.11+11、reactor-core/reactor-test 3.8.8（由 Boot4.0.9 parent/BOM 解析，即 reactor-bom 2025.0.8 线）、reactive-streams 1.0.4、JUnit5；本机macOS arm64。纯JVM内异步实验：无网络IO、无监听端口、无外部服务依赖。 八组测试类回答 F07 Reactor 核心分段的一个问题：冷/热（ColdHotTest）、request(n)契约（RequestContractTest）、背压策略（BackpressureStrategiesTest）、调度器与阻塞隔离（SchedulersBoundariesTest）、错误与重试（RetryBackoffTest）、取消与释放（CancellationCleanupTest）、Context（ContextPropagationTest）、无界堆积反例（UnboundedSlowConsumerTest）。

从仓库根运行完整公开入口：

实验代码

JAVA_HOME=$(/usr/libexec/java_home -v21) python3 labs/reactor-core/run.py

需要Python3、Maven、本机JDK21。每个命令600秒上限；入口一次调用完成全部测试→依赖树核对→evidence写入。原始日志在 ~/.cache/software-engineering-study/java-depth-impl/u26/runtime/，短产物为evidence。

断言表（全部为真实JUnit断言，非日志判读）

组	实际断言与观察
冷/热	冷流两次订阅各自完整执行（副作用计数6=3×2）；Sinks.many().replay().all() 后到订阅者收全部历史 [1,2,3,4]；multicast().directBestEffort() 订阅前 emit 返回 FAIL_ZERO_SUBSCRIBER（值丢弃）、只收订阅后 [2,3]；publish() 冷源只订阅不执行（0次）、connect() 才触发、publish（非replay）的后到者拿不到历史
request(n)	BaseSubscriber 手动 request(2) 订 range(1,10)：恰好2个onNext、0次complete；take(2)：源头只产生2个（emitted=2）且向上游 cancel=1；flatMap 预取：下游只要2、向上游累计请求256；StepVerifier 初始2→补3→取消的分步请求验证
背压策略	interval(2ms)+请求20不补：Drop received=20/dropped≈232（请求量之外全丢）；Latest（publishOn(single,1)解耦交付）收 [0,101,204,307]（忙等200ms期间的最新值，跳号）；Buffer 收到值连续无跳号（堆积在缓冲不丢）；Error received=5后 reactor.core.Exceptions$OverflowException
调度器	publishOn边界：前=main、后=boundedElastic-1；subscribeOn：源头=u26-sub-N、无subscribeOn=main；组合：source=u26-sub（subscribeOn管订阅/源头段）、mid=boundedElastic（publishOn管其下游交付段）；Flux.parallel(2).runOn(parallel) 线程全在 parallel-N
阻塞隔离	parallel线程上 Mono.just("x").map(v->v).block() 抛 IllegalStateException: block()/blockFirst()/blockLast() are blocking, which is not supported in thread parallel-7；实测边界：Mono.just("x").block()（无中间operator）不抛——MonoJust覆写block()直接返回标量值（scalarBlockBypass=true）；boundedElastic承载真实sleep 150ms（elapsed≥140ms）且block在main上调用正常
错误/重试	onError终止流：[next:1, error:mid-failure]（第3个值不再产生、无complete）；onErrorResume分支：[next:1, resume:…, next:99, complete]；Retry.backoff(5,200ms)：3次尝试、两次重试间隔实测276ms/315ms（jitter=0.5→[200,300)ms）、总耗时606ms；耗尽：attempts=3、RetryExhaustedException ← IllegalStateException(boom-3)
取消/释放	dispose()后无限interval计数冻结（frozenAt=10,afterWait=10）；Flux.using cancel时清理执行（cleanupThread=main）；矩阵：手动dispose=(cancel=1,finally=CANCEL)、take收满上游=(cancel=1,finally=CANCEL)/take下游=(cancel=0,finally=ON_COMPLETE)、错误=(cancel=0,finally=ON_ERROR)；timeout(50ms)慢源：fallback交付且upstreamCancelled=1；flatMap inner无限流被take(4)全部取消（activeInners=0）
Context	contextWrite 之前声明的operator读得到（value）、之后声明的读不到（absent）——Context自订阅者向上游传播；跨map/flatMap链传递 a-1-2:v26；subscribeOn切线程后Context仍可见（carried）而ThreadLocal为null
反例	快源+慢消费+无界缓冲（publishOn解耦）：pending=236堆积可观察、dispose后emitted冻结在258（可清理）；对照（无缓冲操作符、请求窗口1）：pending=0——慢消费把tick压到消费速率（上游被拖住），堆积只在有缓冲解耦时发生

关键API核实（源码核对，v3.8.8）

• Flux/Mono 上没有 doOnDispose 操作符：Disposable.dispose() 在订阅链上的可观察信号是 doOnCancel 与 doFinally 的 CANCEL 类型（SignalType 枚举为 ON_COMPLETE/ON_ERROR/CANCEL）。
• onBackpressureFatal 不存在：最接近的边界是 onBackpressureError()（实现为 onBackpressureDrop(t -> { throw Exceptions.failWithOverflow(); })，抛 OverflowException）与 onBackpressureBuffer(maxSize, BufferOverflowStrategy)（ERROR/DROP_LATEST/DROP_OLDEST）。
• block() 检测的精确边界：MonoJust.block()/MonoCallable.block() 直接返回标量值绕过 BlockingSingleSubscriber.blockingGet 的 NonBlocking 线程检测；任何中间operator（map/hide）恢复检测。生产代码不能因“有时不抛”把 block 用在 parallel 上。

只重跑受影响的组

完整入口默认重建并跑全部（全部测试→依赖树→evidence）。测试类之间没有共享状态：只改某个组可单独 mvn -q test -Dtest=<ClassName> 后用run.py同参数重跑；evidence只在完整入口下重写，不把分段运行拼成一次调用。

未运行

真实网络IO流（Reactor Netty）、WebFlux/WebClient/R2DBC链路（DispatcherHandler、注解与函数端点、响应式安全与事务资源）、HTTP层流控与Servlet/reactive两套边界对照——归 #64（U27）及后续票；RSocket归F10票。本票纯JVM内结论不能证明这些范围。重试的虚拟时间测试（StepVerifier.withVirtualTime）未使用：重试间隔以真实墙钟时间验证。