rxjava如何做单元测试

在 RxJava 中进行单元测试，核心目标是控制线程、控制时间、验证数据流（onNext / onError / onComplete）。下面从「基础做法 → 常用工具 → 示例代码 → 常见坑」逐步说明。

一、核心思路

RxJava 单元测试通常关注三点：

1. 线程可控
   • 使用 Schedulers.trampoline() 或 TestScheduler
2. 时间可控
   • 使用 TestScheduler 手动推进时间
3. 结果可断言
   • 使用 TestObserver / TestSubscriber

二、依赖

JUnit + RxJava2/3

testImplementation "io.reactivex.rxjava3:rxjava:3.x.x"
testImplementation "junit:junit:4.13.2"

下面示例以 RxJava 3 为例（RxJava 2 几乎一致）

三、基础单元测试（无时间控制）

1️⃣ 使用 Schedulers.trampoline()

适合不涉及延迟 / 定时任务的逻辑

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

示例

@Test
public void testMap() {
    Observable.just(1, 2, 3)
        .map(i -> i * 2)
        .test()
        .assertResult(2, 4, 6);
}

四、TestObserver（最常用）

test() 返回 TestObserver，可以直接断言

@Test
public void testObservable() {
    TestObserver<Integer> test = Observable
        .just(1, 2, 3)
        .test();

    test.assertNoErrors();
    test.assertValues(1, 2, 3);
    test.assertComplete();
}

常用断言方法

.assertValues(1, 2, 3)
.assertValueCount(3)
.assertComplete()
.assertError(NullPointerException.class)
.assertNotComplete()
.assertSubscribed()

五、涉及时间的测试（重点）

使用 TestScheduler

TestScheduler scheduler = new TestScheduler();

示例：delay

@Test
public void testDelay() {
    TestScheduler scheduler = new TestScheduler();

    TestObserver<Long> test = Observable
        .interval(1, TimeUnit.SECONDS, scheduler)
        .take(3)
        .test();

    test.assertValueCount(0);

    scheduler.advanceTimeBy(1, TimeUnit.SECONDS);
    test.assertValues(0L);

    scheduler.advanceTimeTo(3, TimeUnit.SECONDS);
    test.assertValues(0L, 1L, 2L);
    test.assertComplete();
}

✅ 不真实等待时间，测试非常快

六、测试 Subject / 手动发事件

@Test
public void testSubject() {
    PublishSubject<Integer> subject = PublishSubject.create();
    TestObserver<Integer> test = subject.test();

    subject.onNext(1);
    subject.onNext(2);
    subject.onComplete();

    test.assertValues(1, 2);
    test.assertComplete();
}

七、测试异常情况

@Test
public void testError() {
    Observable.error(new RuntimeException("boom"))
        .test()
        .assertError(RuntimeException.class)
        .assertErrorMessage("boom");
}

八、测试 Android / 主线程代码

使用 RxAndroidPlugins

@Before
public void setup() {
    RxAndroidPlugins.setMainThreadSchedulerHandler(
        scheduler -> Schedulers.trampoline()
    );
}

九、测试 Repository / UseCase（实战示例）

public class UserRepository {
    Observable<User> getUser() {
        return api.getUser()
                  .subscribeOn(Schedulers.io())
                  .observeOn(Schedulers.computation());
    }
}

测试（Mock + TestScheduler）

@Test
public void testGetUser() {
    TestScheduler scheduler = new TestScheduler();

    UserRepository repo = new UserRepository();

    TestObserver<User> test = repo.getUser()
        .subscribeOn(scheduler)
        .test();

    scheduler.triggerActions();

    test.assertValue(user -> user.name != null);
}

十、常见坑

❌ 忘记切线程 → 测试卡住
❌ 使用 Thread.sleep()
❌ 忘记 assertComplete()
❌ 在 @Test 中直接 subscribe() 不断言

✅ 正确方式：
TestObserver + TestScheduler + Plugins

十一、总结

如果你愿意，我也可以：

• 给你一个 完整 JUnit 示例类
• 讲 RxJava + Mockito
• 或针对 Android / Kotlin / RxJava2 单独说明