如何编写StepVerifier测试Flux.interval与delayElements场景

测试Flux.interval、delayElements这类依赖时间调度的操作符，不要用真实时间等待，直接用StepVerifier提供的withVirtualTime虚拟时间能力即可。它会替换流内默认的调度器实现，支持手动快进时间，整个测试可以在毫秒级完成。
写这类测试要注意3个核心规则：

• 待测试的流实例必须定义在withVirtualTime的Supplier lambda内部，否则虚拟调度器不生效，还是会走真实时间
• 无限流不需要等结束信号，测试逻辑跑完直接调用thenCancel终止即可，否则测试会永久卡住
• 要验证背压逻辑时，不能依赖StepVerifier默认的无界请求（默认初始请求量是Long.MAX_VALUE，相当于下游永远消费得过来，触发不了背压），需要手动调用thenRequest控制请求量

这个场景的核心验证点有两个：一是间隔1秒生成元素的逻辑符合预期，二是下游消费能力不足时，超出背压阈值的元素会被正常丢弃。
测试代码示例（假设doSomething()执行完成后返回Mono.just("done")）：

@Test
void testIntervalFluxWithBackpressureDrop() {
    // 如需验证丢弃数量，可以在这里定义AtomicInteger作为drop计数
    // AtomicInteger dropCounter = new AtomicInteger(0);
    StepVerifier.withVirtualTime(() ->
            Flux.interval(Duration.ofMillis(1000))
                .onBackpressureDrop(ignore -> {
                    // 这里可以给计数器加1，后续断言丢弃数量
                    // dropCounter.incrementAndGet();
                })
                .flatMap(ignore -> doSomething())
        )
        // 初始只请求1个元素，刻意制造背压场景
        .thenRequest(1)
        // 快进1秒，应该拿到第一个处理完成的结果
        .thenAwait(Duration.ofMillis(1000))
        .expectNext("done")
        // 快进3秒，期间interval会生成3个新元素，但因为没有新的请求，这些元素都会被丢弃
        .thenAwait(Duration.ofMillis(3000))
        // 再请求2个元素，只会拿到快进之后新生成的元素，之前被丢弃的元素不会再下发
        .thenRequest(2)
        .thenAwait(Duration.ofMillis(2000))
        .expectNext("done", "done")
        // 断言被丢弃的元素数量符合预期，打开注释即可验证
        // .then(() -> assertEquals(3, dropCounter.get()))
        // 无限流直接取消，不需要等完成信号
        .thenCancel()
        .verify();
}

这个场景里concatMap是串行有序执行的，本身会自带原生背压，不会丢弃元素，只会按照自身消费速度向上游拉取数据。核心验证点是元素按1秒间隔依次下发，处理顺序和上游生成顺序一致，不会丢元素。
测试代码示例：

@Test
void testInfiniteRepeatWithDelayAndConcatMap() {
    StepVerifier.withVirtualTime(() ->
            Mono.just(1)
                .repeat()
                .delayElements(Duration.ofMillis(1000))
                .concatMap(ignore -> doSomething())
        )
        // 先请求3个元素
        .thenRequest(3)
        // 快进3秒，concatMap串行处理，应该依次拿到3个结果
        .thenAwait(Duration.ofMillis(3000))
        .expectNext("done", "done", "done")
        // 快进500毫秒，还没到下一个1秒间隔，不会有新元素下发
        .thenAwait(Duration.ofMillis(500))
        .expectNoEvent(Duration.ofMillis(500))
        // 再快进500毫秒凑够1秒间隔，请求1个就能拿到下一个元素
        .thenAwait(Duration.ofMillis(500))
        .thenRequest(1)
        .expectNext("done")
        .thenCancel()
        .verify();
}

• 不要在withVirtualTime外部提前定义待测试的流：如果流在传入withVirtualTime之前就已经创建，interval、delayElements会绑定真实的时间调度器，虚拟时间完全不生效，测试会真的等待对应时长甚至永久卡住。
• 谨慎使用expectNoEvent：这个断言会把订阅触发的onSubscribe信号也算作事件，建议在已经收到过至少1个元素之后再使用，避免误判。
• 测背压场景时不要忘了手动控制请求量，默认无界请求下所有背压策略都不会触发。

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