concatMap 设了零预取，下游取两项，上游为何执行了三次

StepVerifier 只请求两个元素，concatMap(..., 0) 下游也只收到两个，放在 concatMap 前面的计数器却变成了 3。若内部 Publisher 带异步延迟，这不一定是下游收到超出 demand 的数据，而可能是外层又被请求了一项。

Reactor Core 3.8.1 的无预取实现仍然需要协调外层输入与内层 Publisher 的生命周期。prefetch = 0 不等于“任何时刻对外层的请求数都与下游已消费数完全相等”。

内层完成后，会请求下一项外层输入

该版本的 FluxConcatMapNoPrefetch.innerComplete() 在活动内层结束后，将状态切回 REQUESTED，并对上游调用 request(1)。下一条外层数据因此可能已经经过 concatMap 前面的 doOnNext，即使下游暂时没有请求下一条内层结果。

这要与背压违规分开：计数器放在操作符上游，记录的是进入映射之前的输入，不是最终发给订阅者的数量。一个内层还可能发出零项或多项，外层输入数本来就不能一般性地等同于下游结果数。

所以，先把观察点的位置标清楚。分别记录外层 request/onNext、内层订阅与完成、下游接收与取消，而不是只凭一个总计数断言 concatMap 丢失了背压。

thenAwait 会让管线继续运行

下面是放在既有 Reactor Test/JUnit 工程中的片段，flux 是待检查的 Publisher：

StepVerifier.create(flux, 2)
    .expectNextCount(2)
    .thenCancel()
    .verify();

如果目的是收到两个元素后立即结束观察，就不要额外在取消前插入一个 thenAwait，再要求上游计数永远停在 2。等待期间内层可以完成，并触发下一条外层请求。

这个片段验证下游消费后取消，不保证任意异步调度下外层副作用恰好执行两次。需要精确比较发生顺序时，使用受控的 Publisher、请求记录和虚拟时间；虚拟时间中的 Publisher 也应在供应函数里创建，使延迟真正采用测试调度器。

不让上游副作用承担精确业务次数

如果 concatMap 前的 doOnNext 已经扣款、提交任务或占用不可恢复资源，取消订阅并不会回滚这些动作。也不能仅靠把 prefetch 设为零，就保证这些副作用严格与下游成功消费次数一致。

先明确业务要限制的是“最多处理两个输入”，还是“最终取两个结果”。前者可在合适的输入边界限制数据范围；后者还要考虑每个输入发出几个结果，以及取消时正在执行的操作如何结束。移动 take(2) 的位置会改变这项语义，不能把它当作通用修补直接加上。

这条现象不要求为了第三次上游观察去更新 Reactor。验证应同时覆盖正常完成、内层空结果、多结果、错误和取消，并检查那些已经开始的业务动作是否有自己的幂等与结束机制。下游只收到两项，并不证明上游从未开始第三项工作。