java

13.1 StepVerifier

reactor-test 的 StepVerifier 深度使用：期望链 assertNext/expectNext/expectComplete、虚拟时间调度器 VirtualTimeScheduler、背压请求量验证、错误信号断言，以及响应式操作符行为测试。

Leeting Yan 2026-11-01 2 分钟阅读 889 字

StepVerifier 简介

StepVerifier 是 reactor-test 提供的响应式流断言工具。它订阅被测试的 Mono/Flux，按步骤验证每个信号。

基础验证

import reactor.test.StepVerifier;

@Test
void testFlux() {
    Flux<String> flux = Flux.just("a", "b", "c");

    StepVerifier.create(flux)
        .expectNext("a")
        .expectNext("b")
        .expectNext("c")
        .expectComplete()
        .verify();
}

verify() 触发实际订阅。如果不调，StepVerifier 不会执行。

简写形式

StepVerifier.create(flux)
    .expectNext("a", "b", "c")  // 一次断言多个
    .expectComplete()
    .verify();

// 或
StepVerifier.create(flux)
    .expectNextCount(3)
    .expectComplete()
    .verify();

错误断言

@Test
void testError() {
    Mono<String> mono = Mono.error(new IllegalArgumentException("bad"));

    StepVerifier.create(mono)
        .expectError(IllegalArgumentException.class)
        .verify();
}

@Test
void testErrorMessage() {
    StepVerifier.create(Mono.error(new RuntimeException("fail")))
        .expectErrorMessage("fail")
        .verify();
}

assertNext：自定义断言

StepVerifier.create(userRepo.findById(1L))
    .assertNext(user -> {
        assertEquals("alice", user.name());
        assertEquals("ACTIVE", user.status());
    })
    .expectComplete()
    .verify();

虚拟时间

测试时间相关操作符时，不需要真的等待：

@Test
void testWithVirtualTime() {
    StepVerifier.withVirtualTime(() ->
            Flux.interval(Duration.ofMinutes(1))
                .take(3)
        )
        .expectSubscription()
        .expectNoEvent(Duration.ofMinutes(1))
        .expectNext(0L)
        .thenAwait(Duration.ofMinutes(1))
        .expectNext(1L)
        .thenAwait(Duration.ofMinutes(1))
        .expectNext(2L)
        .expectComplete()
        .verify();
}

withVirtualTime(Supplier)：在虚拟时间调度器上创建流，测试本身不用真的等 3 分钟
expectNoEvent(Duration)：断言该时间段内没有任何信号；订阅本身算一个事件，所以要先用 expectSubscription()
thenAwait(Duration)：快进虚拟时间

不要在测试里先手动 VirtualTimeScheduler.getOrSet() 再调用 withVirtualTime：前者会替换全局调度器，后者自己管理虚拟时间，混用容易让后续测试继承一个已被推进的全局时钟。

背压验证

@Test
void testBackpressure() {
    StepVerifier.create(
            Flux.range(1, 100),
            0  // 初始请求 0 个
        )
        .expectSubscription()
        .thenRequest(1)
        .expectNext(1)
        .thenRequest(2)
        .expectNext(2, 3)
        .thenRequest(97)
        .expectNextCount(97)
        .expectComplete()
        .verify();
}

StepVerifier.create(publisher, initialRequest) 控制初始 request(n)。

thenCancel：取消订阅

@Test
void testCancellation() {
    StepVerifier.create(Flux.interval(Duration.ofMillis(100)))
        .expectNext(0L)
        .expectNext(1L)
        .thenCancel()
        .verify();  // 仍然要 verify：它会执行订阅并确认取消发生在预期位置
}

适用于无限流或需要验证取消行为的场景。thenCancel() 只声明「到这里取消」，不调用 verify() 同样不会真正跑测试。

验证超时

StepVerifier.create(slowMono)
    .expectNext("result")
    .expectComplete()
    .verify(Duration.ofSeconds(5));  // 总验证超时 5 秒

如果 slowMono 超过 5 秒未完成，verify(Duration) 抛 AssertionError。

测试事件流处理

@Test
void testEventProcessing() {
    EventProcessor processor = new EventProcessor();

    StepVerifier.create(processor.processEvents())
        .expectNextMatches(event -> event.type().equals("CREATED"))
        .expectNextMatches(event -> event.type().equals("UPDATED"))
        .expectNextMatches(event -> event.type().equals("DELETED"))
        .expectComplete()
        .verify();
}

expectNextMatches(Predicate) 用于复杂条件的断言。

StepVerifier 在 WebFlux 测试中的角色

WebFlux 的 WebTestClient 已经封装了 HTTP 层面的断言。StepVerifier 用于测试纯响应式逻辑：

// 测试 Repository 返回的 Flux
StepVerifier.create(orderRepo.findByStatus("PENDING"))
    .expectNextCount(10)
    .expectComplete()
    .verify();

// 测试 Service 层的转换逻辑
StepVerifier.create(orderService.enrichOrders(orders))
    .assertNext(order -> assertNotNull(order.userName()))
    .expectNextCount(9)
    .expectComplete()
    .verify();

小结

API	用途
`expectNext(T...)`	断言下一个元素
`expectNextCount(n)`	断言 n 个元素（不检查值）
`expectNextMatches(Predicate)`	断言元素满足条件
`assertNext(Consumer)`	自定义断言
`expectComplete()`	断言正常完成
`expectError(Class)`	断言错误类型
`expectNoEvent(Duration)`	断言时间段无信号
`thenAwait(Duration)`	快进虚拟时间
`thenRequest(n)`	请求 n 个元素
`thenCancel()`	取消订阅
`verify(Duration)`	执行验证，带超时

下一节 13.2 Testcontainers ：PostgreSQL/Redis 容器集成测试、R2DBC 连接与数据初始化。

继续阅读

探索更多技术文章

浏览归档，发现更多关于系统设计、工具链和工程实践的内容。

全部文章返回首页