测试RxJava的Single.zip()：确保所有参数均被执行

哈哈，这个问题我之前测试RxJava的时候也踩过一模一样的坑！本质上是RxJava的异步调度器和测试同步逻辑的冲突，我来给你拆解清楚：

首先先看你的核心代码：

public Completable execute(String id) {
    return doThing()
        .flatMap(result -> Single.zip(
            employeeService.getEmployee(id),
            databaseService.getData(id),
            (employee, data) -> ...
        ))
        .toCompletable(...);
}

为什么await没起到预期的阻塞效果？ #

你以为await是阻塞式的，但忽略了RxJava默认的异步调度器——比如employeeService或databaseService的方法大概率用了Schedulers.io()这类后台线程执行任务。测试主线程调用await后，后台任务还在跑，主线程却可能已经走到验证逻辑，自然会出现验证失败的情况。

具体解决步骤 #

1. 替换测试环境的RxJava调度器

在测试类中，用RxJava的插件把所有异步调度器替换为同步执行的TrampolineScheduler，这样所有任务都会在当前线程按顺序执行，await就能真正阻塞到任务完成：

@Before
public void setupTest() {
    // 替换IO、计算等调度器为同步执行
    RxJavaPlugins.setIoSchedulerHandler(scheduler -> Schedulers.trampoline());
    RxJavaPlugins.setComputationSchedulerHandler(scheduler -> Schedulers.trampoline());
}

@After
public void cleanupTest() {
    // 测试结束后重置插件，避免影响其他测试用例
    RxJavaPlugins.reset();
}

2. 合理设置await的超时时间

如果你的测试用了带超时的await(timeout, timeUnit)，超时时间太短会导致任务还没执行完就触发超时，测试提前结束。建议先设置一个足够长的超时（比如10秒），或者在本地测试时用无参await()（注意CI环境慎用，避免无限阻塞）。

3. 确认Single.zip的订阅逻辑

这里要注意RxJava Single.zip的特性：只要其中一个Single发射错误事件，整个zip会立即终止并发射错误，但如果其中一个Single是同步抛出错误（比如mock时直接返回Single.error()且没有调度器切换），另一个Single可能根本不会被订阅，自然不会被调用。

这种情况下，你需要确保mock的服务方法是异步发射错误，比如：

when(employeeService.getEmployee(anyString()))
    .thenReturn(Single.error(new RuntimeException("test error"))
    .subscribeOn(Schedulers.io())); // 强制异步，确保zip会订阅两个Single

4. 测试代码要正确等待Completable完成

测试时一定要等Completable执行完成后再做验证，比如：

@Test
public void testExecuteWithFailure() {
    // 准备mock数据
    when(employeeService.getEmployee("testId")).thenReturn(Single.error(new RuntimeException("oops")));
    when(databaseService.getData("testId")).thenReturn(Single.just(mockData()));

    // 执行方法并等待完成
    testSubject.execute("testId")
        .await(10, TimeUnit.SECONDS)
        .assertError(RuntimeException.class); // 验证错误结果

    // 现在再验证两个服务方法都被调用
    verify(employeeService).getEmployee("testId");
    verify(databaseService).getData("testId");
}

额外小技巧（RxJava 3适用） #

如果需要更精细地控制异步任务的执行时机，可以用TestScheduler：

TestScheduler testScheduler = new TestScheduler();

@Before
public void setup() {
    RxJavaPlugins.setIoSchedulerHandler(scheduler -> testScheduler);
}

@Test
public void testExecuteWithTestScheduler() {
    TestObserver<Void> testObserver = testSubject.execute("testId").test();
    
    // 手动触发所有异步任务执行
    testScheduler.triggerActions();
    
    // 验证结果和方法调用
    testObserver.assertError(RuntimeException.class);
    verify(employeeService).getEmployee("testId");
    verify(databaseService).getData("testId");
}

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