使用Scheduler时StepVerifier.verifyComplete()陷入死循环问题排查

我在测试基于Reactor Mono的编排流程时遇到两个问题：

• 使用StepVerifier测试时，测试始终无法结束，但移除publishOn(myThreadDispatcher)后测试就能正常完成。
• 尝试用StepVerifier.withVirtualTime解决，结果线程过早结束，直接返回onComplete而非预期的onNext。

请指出代码中需要添加或修正的内容：

业务代码 #

@Component
public class MyOrchestrator {

    @Autowired
    @Qualifier("myThreadDispatcher")
    private Scheduler myThreadDispatcher;

    @Autowired
    private PrepareRequestInvoker prepareRequestInvoker;

    @Autowired
    private EventHandler eventHandler;

    @Override
    public Mono<Result> orchestrate(FooRequest fooRequest) {
        return Mono.just(fooRequest)
                .publishOn(myThreadDispatcher)
                .flatMap(request -> prepareRequestInvoker.invoke(request))
                .flatMap(queryResult -> eventHandler.orchestrate(mockResult, fooRequest));
    }

}

测试代码 #

@RunWith(PowerMockRunner.class)
@PowerMockIgnore({"javax.management.*", "javax.crypto.*"})
class MyTest{

    @Mock
    private Scheduler myThreadDispatcher;

    @Mock
    private PrepareRequestInvoker prepareRequestInvoker;

    @Mock
    private EventHandler eventHandler;

    @InjectMocks
    private MyOrchestrator orchestrator;

    Result mockResult = Result.from("test");
    Mono<Result> resultMono = Mono.fromCompletionStage(CompletableFuture.completedFuture(mockResult));

    @Test
    public void testMono(){
        prepareTest(createRequest());
        Mockito.when(prepareRequestInvoker.invoke(fooRequest)).thenReturn(resultMono);
        Mockito.when(eventHandler.orchestrate(mockResult,fooRequest)).thenReturn(resultMono);
        Mono<Result> monoToVerify= orchestrator.orchestrate(fooRequest);

        StepVerifier.withVirtualTime(monoToVerify)
                .expectNext(Result.from("test"))
                .verifyComplete();
    }

}

解决方案 #

问题核心 #

你Mock的myThreadDispatcher没有实际执行任务的逻辑，publishOn将任务提交给这个Mock调度器后，任务永远不会被执行，导致Mono流程卡住；而用虚拟时间时，自定义调度器未绑定虚拟时间，虚拟时间无法驱动任务执行，所以提前触发onComplete。

修正方案1：使用测试专用真实调度器 #

直接替换Mock的调度器为Reactor提供的测试用调度器，比如Schedulers.immediate()（同步执行任务，适合测试）：

// 移除@Mock注解，改用真实调度器
private Scheduler myThreadDispatcher = Schedulers.immediate();

如果需要模拟异步场景，可改用Schedulers.boundedElastic()，测试时无需额外操作，Reactor会自动处理任务执行。

修正方案2：正确Mock调度器的执行逻辑 #

如果必须Mock调度器，需要让它的Worker立即执行提交的任务：

@Before
public void setup() {
    Worker mockWorker = Mockito.mock(Worker.class);
    // 让调度器的Worker立即执行任务
    Mockito.when(mockWorker.schedule(Mockito.any(Runnable.class))).thenAnswer(invocation -> {
        Runnable task = invocation.getArgument(0);
        task.run();
        return Disposable.empty();
    });
    Mockito.when(myThreadDispatcher.createWorker()).thenReturn(mockWorker);
}

这样publishOn提交的任务会被立即执行，流程不会卡住。

修正方案3：正确使用VirtualTime #

要用虚拟时间控制所有异步流程，需确保publishOn使用虚拟时间调度器：

@Test
public void testMonoWithVirtualTime(){
    prepareTest(createRequest());
    Mockito.when(prepareRequestInvoker.invoke(fooRequest)).thenReturn(resultMono);
    Mockito.when(eventHandler.orchestrate(mockResult,fooRequest)).thenReturn(resultMono);
    
    // 使用虚拟时间调度器替换原调度器
    VirtualTimeScheduler virtualScheduler = VirtualTimeScheduler.get();
    orchestrator.myThreadDispatcher = virtualScheduler;

    // 注意：withVirtualTime要接收Supplier，确保流程使用虚拟时间
    StepVerifier.withVirtualTime(() -> orchestrator.orchestrate(fooRequest))
            .thenAwait(Duration.ofMillis(100)) // 推进时间触发任务执行
            .expectNext(Result.from("test"))
            .verifyComplete();
}

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