测试中使用Awaitility等待WebClient远程Flux初始化的问题

API 探路者

2026-8-4

问题根源与解决方案

首先要明确：你误解了Reactor中Flux的工作机制——getFlux方法返回的Flux对象本身是立即创建的，不存在"初始化未完成"的情况。但Flux是冷流，只有当你调用subscribe()之后，才会真正发起HTTP请求并开始接收数据。你之前用Awaitility等待events != null完全无效，因为events = getFlux(guid)执行完毕后，events就已经是非空的Flux对象了，可此时HTTP请求还没触发，自然会出现竞态。

测试提前结束的核心原因是：测试主线程没有等待订阅后的异步数据流完成，就直接终止了，导致远程返回的测试数据还没被处理，测试就结束了。

正确的测试方案

方案1：用block方法等待流完成（适合有限流）

如果你的测试接口返回的是有限数量的事件，直接用block系列方法让主线程阻塞等待流结束：

// 替换原有的subscribe和Awaitility代码
// 收集所有事件并等待最多5秒
List<String> receivedEvents = events.collectList().block(Duration.ofSeconds(5));
// 可添加断言验证结果，比如：
Assert.assertEquals(预期事件数量, receivedEvents.size());
Assert.assertTrue(receivedEvents.contains("预期的事件内容"));

如果只需要等待第一个事件，用blockFirst：

String firstEvent = events.blockFirst(Duration.ofSeconds(5));
Assert.assertNotNull(firstEvent);

方案2：用CountDownLatch手动控制等待（适合无限流）

如果接口返回的是无限流（不会主动complete），可以用CountDownLatch来等待指定数量的事件：

// 假设需要等待1个事件
CountDownLatch latch = new CountDownLatch(1);

events.subscribe(
    event -> {
        Logs.Info("event: " + event);
        latch.countDown(); // 收到事件后释放 latch
    },
    error -> {
        Logs.Error("订阅出错", error);
        latch.countDown(); // 出错时也释放，避免测试挂起
    }
);

// 等待最多5秒，超时则断言失败
if (!latch.await(5, TimeUnit.SECONDS)) {
    Assert.fail("5秒内未收到预期事件");
}

方案3：用Reactor Test工具（推荐用于单元测试）

如果是编写单元测试，推荐使用Reactor官方的reactor-test依赖，用StepVerifier来验证响应式流：

StepVerifier.create(events)
    .expectNext("第一个预期事件") // 可添加多个expectNext匹配事件序列
    .expectNext("第二个预期事件")
    .expectComplete() // 如果是有限流，预期流最终完成
    .timeout(Duration.ofSeconds(5))
    .verify();

StepVerifier会自动阻塞主线程，直到流满足所有预期条件或超时，无需手动处理等待逻辑。

额外注意事项

你的代码存在类型不匹配问题：初始定义的是private Flux<Event> events = null;，但getFlux返回的是Flux<String>，编译时会报错，需要统一类型。
onStatus中返回Mono.empty()会让流在遇到401状态时直接静默结束，不会抛出错误，测试时如果需要感知这种情况，建议改成return Mono.error(new UnauthorizedException())，方便测试捕获。

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