测试RxJava2_Java_Andres Almiray

【AICon】如何构建高效的 RAG 系统？RAG 技术在实际应用中遇到的挑战及应对策略？>>> 了解详情 



 写点什么

译者注：关注 InfoQ 资讯的读者可能已经留意到，我们前面给出了一篇很好的 RxJava 文章“测试 RxJava ”。本文是上一篇文章的修订，用相同的例子循序渐进地介绍了如何测试 RxJava2。译者将两篇文章中的不同之处用粗体标识出来，并使用添加注释的形式说明示例代码中的差异之处，以供读过前篇的读者快速浏览本文。

关键要点：

RxJava 含有内建的、测试友好的解决方案。

使用 TestSubscriber 去验证 Observable。

使用 TestScheduler 可实现对时间的严格控制。

Awaitility 库提供了对测试环境进一步的控制。

本文是“测试RxJava ”一文的修订，根据RxJava2 规范做了全面更新。

你已经阅读过RxJava 的相关内容，也已经在互联网上体验过像“ RxJava 实例解析”中的那些示例，现在打算在自己的代码中探索一下响应式编程了。但是，现在却一直困扰着如何测试那些可能会在代码库中发现的新功能呢？

使用响应式编程，就必须转变对给定问题的推理方式，因为我们要聚焦于作为事件流的流动数据，而非个别数据项。事件通常是被不同的线程所产生和消费，因此在编写测试时必须要对并发问题有着清晰的认识。幸运的是，RxJava 提供了测试 Observable 和 Subscription 的内建支持，并且是直接构建于 RxJava 的核心依赖中。

第一步

让我们回顾一下在“ RxJava by Example ”一文中所给出的那个词汇的例子，看一下如何对该例子作测试。让我们从基础测试工具的设置开始。在我们的测试架构中，使用了 JUnit 作为测试工具。

复制代码

 import io.reactivex.Observable;
import io.reactivex.observers.TestObserver;
import io.reactivex.plugins.RxJavaPlugins;
import io.reactivex.schedulers.Schedulers;
//RxJava2 中，包名由 rx.xxx 改为 io.reactivex.xxx。
import org.junit.Test;
import java.util.*;
import static java.util.concurrent.TimeUnit.SECONDS;
import static org.awaitility.Awaitility.await;
import static org.junit.Assert.assertThat;
import static org.hamcrest.Matchers.*;
 
 
public class RxJavaTest {
    private static final List<String> WORDS = Arrays.asList(
       "the",
       "quick",
       "brown",
       "fox",
       "jumped",
       "over",
       "the",
       "lazy",
       "dog"
    );
}

事实上在没有给定调度器（Scheduler）的情况下，Subscription 将默认运行于调用线程上。因此我们将在首个测试中使用原生的方法。这意味着我们可实现一个 Subscription 接口的对象，在 Subscription 发生后就立刻对其状态做断言（assert）。

复制代码

 @Test
public void testInSameThread() {
   // given:
   List<String> results = new ArrayList<>();
   //Observable 的 from 方法改为 fromIterable。
   Observable<String> observable = Observable.fromIterable(WORDS) 
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string));
 
   // when:
   observable.subscribe(results::add);
 
   // then:
   assertThat(results, notNullValue());
   assertThat(results, hasSize(9));
   assertThat(results, hasItem(" 4. fox"));
}

注意这里使用了显式的 List容器，与实际订阅者一起累计结果。由于给定的测试很简单，所以可能会使你认为这种显式累加器的方法已经足够好了。但是切记产品级的 Observable 中可能封装了错误或可能产生意外的事件。例子中的 Subscriber 与累加器的简单组合并不足以覆盖这种情况。但不用为此烦恼，RxJava 提供的 TestSubscriber 类型就是用于处理这种情况的。下面我们使用 TestSubscriber 类型重构上面的测试。

复制代码

 @Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingTestObserver() {
   // given:
   TestObserver<String> observer = new TestObserver<>();
   //Observable 的 from 方法改为 fromIterable。
   Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string));
 
   // when:
   observable.subscribe(observer);
 
   // then:
   observer.assertComplete();
   observer.assertNoErrors();
   observer.assertValueCount(9);
   //getOnNextEvents 方法该为 values 方法。
   assertThat(observer.values(), hasItem(" 4. fox"));
}

TestObserver 不仅可替代用户累加器，还另给出了一些行为。例如它能够给出接收到的消息和每个事件相关数据的规模，它也可对 Subscription 被完成且在 Observable 消费期间没有错误出现的状态做断言。虽然当前测试中的 Observable 并未生成任何的错误，但是回到“ RxJava by Example ”一文，我们从中得知了 Observable 将例外与数据事件等同对待。我们可通过如下的方式通过连接例外事件而模拟错误：

复制代码

 @Test
// 例子中所有的 Subscriber 改为 Observer。
public void testFailure() {
   // given:
   TestObserver<String> observer = new TestObserver<>();
   Exception exception = new RuntimeException("boom!");
 
    //Observable 的 from 方法改为 fromIterable。
    Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string))
       .concatWith(Observable.error(exception));
 
   // when:
   observable.subscribe(observer);
 
   // then:
   observer.assertError(exception);
   observer.assertNotComplete();
}

在我们所给出的有限用例中，所有的机制运行良好。但是实际的产品代码可能会完全不同于例子。因此在下文中，我们将考虑一些更加复杂的产品实例。

定制调度器（Scheduler）

在产品代码中，很多用例中的 Observable 都是在特定的线程上执行，这种线程在响应式编程环境中被称为“调度器（Scheduler）”。很多 Observable 操作将可选的调度器参数作为附加参数使用。RxJava 定义了一系列任何时候都可用的命名调度器，包括 IO 调度器（io)、计算调度器（computation，为共享线程）和新线程调度器（newThread）。开发人员也可去实现个人定制的调度器。让我们通过指定计算调度器来修改 Observable 的代码吧。

复制代码

 @Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingComputationScheduler() {
   // given:
   TestObserver<String> observer = new TestObserver<>();
   //Observable 的 from 方法改为 fromIterable。
   Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string));
 
   // when:
   observable.subscribeOn(Schedulers.computation())
       .subscribe(observer);
 
   // 修订版中新添加语句。
   await().timeout(2, SECONDS)
       .until(observer::valueCount, equalTo(9));
 
   // then:
   observer.assertComplete();
   observer.assertNoErrors();
   assertThat(observer.values(), hasItem(" 4. fox"));
}

当运行时就会立刻发现该代码是存在问题的。Subscriber 在测试线程上执行其断言，但是 Observable 在后台线程（计算线程）上生成值。这意味着执行 Subscriber 断言可能先于 Observable 生成所有相关事件，因而导致测试的失败。

为使测试顺利执行，有如下的一些策略可选：

将 Observable 转化为阻塞式的。
强制测试等待，直至给定的条件被满足。
将计算调度器转换为即刻（Schedulers.immediate()）调度器。

我们将对每个策略做展开介绍，但将从“将 Observable 转化为阻塞式”开始，因为实现该策略所需做的技术工作最少，这些工作与所使用的调度器无关。我们假设数据在后台线程中生成，这将导致 Subscriber 从同一后台线程得到通知。

我们要做的是强制生成所有的事件，并在下一个声明被执行前就在测试中完成 Observable。这是通过在 Observable 自身上调用blockingIterable()方法实现的。

复制代码

 @Test
public void testUsingBlockingCall() {
   // given:
   //Observable 的 from 方法改为 fromIterable。
   Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string));
 
   // when:
   //RxJava2 中，toBlocking() 和 toIterable() 方法改为 blockingIterable()
   Iterable<String> results = observable
       .subscribeOn(Schedulers.computation())
       .blockingIterable();
 
   // then:
   assertThat(results, notNullValue());
   assertThat(results, iterableWithSize(9));
   assertThat(results, hasItem(" 4. fox"));
}

该方法虽然适用于我们所给出的简单代码，但可能并不适用于实际的产品代码。如果生产者生成所有的数据需要很长的时间，那将会产生什么后果？这将使测试变得非常慢，并增加了编译时间，还可能会有其它的性能问题。幸运的是，TestObserver 提供了一系列方法，强制测试等待事件的结束。下面给出了一种实现方法：

复制代码

 // 修订版中新添加的代码段。
@Test
public void testUsingComputationScheduler() {
// given:
   TestObserver<String> observer = new TestObserver<>();
   Observable<String> observable = Observable.fromIterable(WORDS)
        .zipWith(Observable.range(1, Integer.MAX_VALUE),
            (string, index) -> String.format("%2d. %s", index, string));
 
    // when:
    observable.subscribeOn(Schedulers.computation())
            .subscribe(observer);
 
    observer.awaitTerminalEvent(2, SECONDS);
 
    // then:
    observer.assertComplete();
    observer.assertNoErrors();
    assertThat(observer.values(), hasItem(" 4. fox"));
}

如果这些方法还不足以满足需求，这里我推荐一个便利的程序库，就是 Awaitility 。简单地说，Awaitility 是一个以精确、简单易读的方式对异步系统相关期望进行表述的 DSL。在项目中可以用 Maven 添加 Awaitility 的依赖关系。

复制代码

 <dependency>
    <groupId>org.awaitility</groupId>
    <artifactId>awaitility</artifactId>
    <version>2.0.0</version>
    <scope>test</scope>
</dependency>

或是使用 Gradle：

testCompile 'org.awaitility:awaitility:2.0.0'Awaitility DSL 的接入点是 org.awaitility.Awaitility.await() 方法（参见下面例子中的第 13 和 14 行代码）。可以使用 Awaitility 定义使测试继续所必须达成的条件，也可在条件中加入超时或其它的时序约束，例如最小、最大或持续范围。对于上面的例子，下面的代码给出了如何在结果中使用 Awaitility：

复制代码

 @Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingComputationScheduler_awaitility() {
    // given:
    TestObserver<String> observer = new TestObserver<>();
    //Observable 的 from 方法改为 fromIterable。
    Observable<String> observable = Observable.fromIterable(WORDS)
        .zipWith(Observable.range(1, Integer.MAX_VALUE),
            (string, index) -> String.format("%2d. %s", index, string));
 
    // when:
     observable.subscribeOn(Schedulers.computation())
         .subscribe(observer);
 
     await().timeout(2, SECONDS)
         .until(observer::valueCount, equalTo(9));
 
     // then:
     observer.assertComplete();
     observer.assertNoErrors();
     //getOnNextEvents() 方法改为 values() 方法。
    assertThat(observer.values(), hasItem(" 4. fox"));
 }

此版本测试并未以任何方式改变 Observable 的本质，这使得你做测试时不必对产品代码做任何改动。该版本测试使用最多 2 秒的等待时间通过检查 Subscriber 状态使 Observable 执行其作业。如果一切进行顺利，在 2 秒内就可将 Subscriber 的状态释放给所有的 9 个事件。

Awaitility 具有和 Hamcrest 的匹配符、Java 8 的 lambda 表达式和方法引用等的良好协作，从而给出精确的和可读的测试条件。Awaitility 还提供了预制扩展，用于那些被广泛使用的 JVM 语言，其中包括 Groovy 和 Scala。

我们要给出最后一个策略中使用了 RxJava 的扩展机制，该扩展是以 RxJava API 的组成部分发布的。RxJava 中定义了一系列的扩展点，允许对几乎任何默认的 RxJava 行为进行微调。这种扩展机制使我们可以针对特定的 RxJava 特性提供修改过的值。利用该机制，在无需关心生成代码中所指定的调度器的情况下，我们可在测试中注入选定的调度器。这正是我们所寻找的方法，该方法被封装在 RxJavaHooks 类中。假设产品代码依赖于计算调度器，我们将覆盖它的默认值，返回一个调度器，它作为被调用的代码使事件处理发生，这是入队调度器（Schedulers.trampoline()）。下面给出测试的代码：

复制代码

 @Test
// 名称中的 Hook 改为 Plugins，代码中所有 subscriber 改为 observer。
public void testUsingRxJavaPluginsWithImmediateScheduler() {
    // given:
    // 调度器由 immediate 改为 trampoline。
    RxJavaPlugins.setComputationSchedulerHandler(scheduler -> 
                                                   Schedulers.trampoline());
     TestObserver<String> observer = new TestObserver<>();
     //Observable 的 from 方法改为 fromIterable。
     Observable<String> observable = Observable.fromIterable(WORDS)
         .zipWith(Observable.range(1, Integer.MAX_VALUE),
             (string, index) -> String.format("%2d. %s", index, string));
 
     try {
         // when:
         observable.subscribeOn(Schedulers.computation())
             .subscribe(observer);
 
         // then:
         observer.assertComplete();
         observer.assertNoErrors();
         observer.assertValueCount(9);
         assertThat(observer.values(), hasItem(" 4. fox"));
     } finally {
         RxJavaPlugins.reset();
     }
}

在测试中，产品代码察觉不到计算调度器是即刻的。请注意钩子函数必须被重置，否则即刻调度器的设置可能会发生泄漏，导致在各处的测试被破坏。使用 try/finall 代码块会在一定程度上模糊了测试的目的，但是幸运的是我们可以使用 JUnit 规则重构该行为，使测试更加精炼，结果更可读。下面给出使用上述规则的一种可能的实现代码：

复制代码

 // 由 Public 改为 private static
private static class ImmediateSchedulersRule implements TestRule {
   @Override
   public Statement apply(final Statement base, Description description) {
       return new Statement() {
           @Override
           // 所有的调度器由 immediate 改为 trampline。
           public void evaluate() throws Throwable {
               RxJavaPlugins.setIoSchedulerHandler(scheduler -> 
                                                Schedulers.trampoline());
               RxJavaPlugins.setComputationSchedulerHandler(scheduler -> 
                                                Schedulers.trampoline());
               RxJavaPlugins.setNewThreadSchedulerHandler(scheduler -> 
                                                Schedulers.trampoline());
 
               try {
                   base.evaluate();
               } finally {
                   RxJavaPlugins.reset();
               }
           }
       };
   }
}

此外，我们还对另外两个调度器的生成方法做了重写。该规则对此后其它的测试目标更为通用。在新的测试用例类中，该规则的使用方法很直接，只需简单地定义一个域，并将其中新类型标注为 @Rule 即可。示例代码如下：

复制代码

 @Rule
public final ImmediateSchedulersRule schedulers = 
                                         new ImmediateSchedulersRule();
 
@Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingImmediateSchedulersRule() {
   // given:
   TestObserver<String> observer = new TestObserver<>();
   //Observable 的 from 方法改为 fromIterable。
   Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.range(1, Integer.MAX_VALUE),
           (string, index) -> String.format("%2d. %s", index, string));
 
   // when:
   observable.subscribeOn(Schedulers.computation())
       .subscribe(observer);
 
   // then:
   observer.assertComplete();
   observer.assertNoErrors();
   observer.assertValueCount(9);
   assertThat(observer.values(), hasItem(" 4. fox"));
}

最终我们可得到与前面测试一样的行为，却没有像前面测试那样的杂乱。下面用一些篇幅来回顾一下我们目前已经做到的事情：

Subscribers 将在同一线程中处理数据，只要没有使用特定的调度器。这意味着在 Subscriber 向 Observable 做订阅后，我们就可在该 Subscriber 上做断言。
TestObserver可累计事件，并给出自身状态的追加断言。
任何对象都可转换为阻塞式的，这使得无论对象使用何种调度器，我们都可以同步等待事件的生成。
RxJava 提供了扩展机制，允许开发人员重写其默认方法，并以适当的方式注入到产品代码中。
并发代码可使用 Awaitility DSL 测试。

上述的每个技术都作用于不同的场景中，但是所有技术都是通过“共同的线程”（译者注：作者在原文中指出 common thread 是作为双关语使用的，其另一个意思是“类似的思路”）相关联：在对 Subscriber 状态做断言之前，测试代码需等待 Observable 完成。考虑到 Observable 的行为会生成数据，是否有方法对该行为进行检查呢？换句话说，是否可以用编程的方式做 Observable 的现场调试？我们将在后文中给出这样的技术。

操控时间

到目前为止我们已用黑箱方式测试了 Observable 和 Subscription。下面我们将考虑另外一种操控时间的技术，该技术使我们可以在 Observable 依然处于活动状态时，打开引擎盖去查看 Subscriber 状态。换句话说，我们将使用采用了 RxJava 的 TestScheduler 类白箱测试技术，这可以说是 RxJava 再一次来救场。这种特定的调度器可精确地设定时间的内部使用方式，例如可将时间提前半秒，或是使时间跳跃 5 秒。我们将首先给出这种新调度器实例的创建方法，然后再讨论代码的测试。

复制代码

  @Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingTestScheduler() {
     // given:
     TestScheduler scheduler = new TestScheduler();
     TestObserver<String> observer = new TestObserver<>();
     Observable<Long> tick = Observable.interval(1, SECONDS, scheduler);
 
     Observable<String> observable = Observable.fromIterable(WORDS)
         .zipWith(tick,
             (string, index) -> String.format("%2d. %s", index, string));
 
     observable.subscribeOn(scheduler)
         .subscribe(observer);
 
     // expect:
     observer.assertNoValues();
     observer.assertNotComplete();
 
     // when:
     scheduler.advanceTimeBy(1, SECONDS);
 
     // then:
     observer.assertNoErrors();
     observer.assertValueCount(1);
     observer.assertValues(" 0. the");
 
     // when:
     scheduler.advanceTimeTo(9, SECONDS);
     observer.assertComplete();
     observer.assertNoErrors();
     observer.assertValueCount(9);
 }

该“产品”代码有了略微的改变，这是由于我们使用了绑定到调度器时隙（interval()）的方法生成计数（第 6 行），而非生成一个计数的范围。但这样做具有一个副作用，就是计数是从零开始生成的，而非从 1 开始。一旦配置了 Observable 和测试调度器，我们立刻做出这样的断言，即假定 Subscriber 不具有值（第 15 行）且没有被完成或生成任何的错误（第 16 行）。这是一个完整性测试，因为此时调度器并没有被移动，因而没有任何值被 Observable 产生或是被 Subscriber 接收到。

下面将时间向前调 1 整秒（第 21 行），该操作将会导致 Observable 生成第一个值，这正是随后的断言集所要检查的（第 24 到 26 行）。

下面将时间从当前时间调到 9 秒。需要注意的是，这意味着将时间准确地调整为调度器启动后的第 9 秒（并非是向前调 1 秒后再向前调 9 秒，即调度器检查启动后的第 10 秒）。换句话说，advanceTimeBy() 方法将调度器的时间调整为相对于当前位置的时间，而 advanceTimeTo() 以绝对的方式调整时间。此后我们做出下一轮的断言（第 30 到 32 行），用于确保所有的数据由 Observable 生成且被 Subscriber 消费。另一件需要说明的事情就是使用 TestScheduler 时，真实的时间是立刻发生调整的，这着意味着测试并不用实际等待 9 秒才去完成。

正如你所看到的，该调度器的使用是非常便利的，仅需将该调度器提供给正在测试的 Observable 即可。但是对使用了指定类型调度器的 Observable，该调度器并不能很好地适用。但是稍等一下，之前我们看到的是如何使用RxJavaPlugins切换一个不影响生产代码的调度器，而这一次是提供一个代替即刻调度器的 TestScheduler。我们甚至可以 apply 定制 JUnit 规则同样的技术，使之前的代码可以用更重用的方式予以重写。首先该新规则为：

复制代码

 // 由 public 改为 private static，所有 Hooks 改为 Plugins。
private static class TestSchedulerRule implements TestRule {
   private final TestScheduler testScheduler = new TestScheduler();
 
   public TestScheduler getTestScheduler() {
       return testScheduler;
   }
 
   @Override
   public Statement apply(final Statement base, Description description) {
       return new Statement() {
           @Override
           public void evaluate() throws Throwable {
               RxJavaPlugins.setIoSchedulerHandler(scheduler -> 
                                                   testScheduler);
               RxJavaPlugins.setComputationSchedulerHandler(scheduler -> 
                                                   testScheduler);
               RxJavaPlugins.setNewThreadSchedulerHandler(scheduler -> 
                                                   testScheduler);
 
               try {
                   base.evaluate();
               } finally {
                   RxJavaPlugins.reset();
               }
           }
       };
   }
}

紧接着是实际的测试代码（在一个新的测试用例类中），去使用我们的测试规则：

复制代码

 @Rule
public final TestSchedulerRule testSchedulerRule = new TestSchedulerRule();
 
@Test
// 例子中所有的 Subscriber 改为 Observer。
public void testUsingTestSchedulersRule() {
   // given:
   TestObserver<String> observer = new TestObserver<>();
 
   Observable<String> observable = Observable.fromIterable(WORDS)
       .zipWith(Observable.interval(1, SECONDS),
           (string, index) -> String.format("%2d. %s", index, string));
 
   observable.subscribeOn(Schedulers.computation())
       .subscribe(observer);
 
   // expect
   observer.assertNoValues();
   observer.assertNotComplete();
 
   // when:
   testSchedulerRule.getTestScheduler().advanceTimeBy(1, SECONDS);
 
   // then:
   observer.assertNoErrors();
   observer.assertValueCount(1);
   observer.assertValues(" 0. the");
 
   // when:
   testSchedulerRule.getTestScheduler().advanceTimeTo(9, SECONDS);
   observer.assertComplete();
   observer.assertNoErrors();
   observer.assertValueCount(9);
}

这样你就成功地实现了它。使用经由 RxJavaHooks 注入 TestScheduler 的方法，可在无需更改原始 Observable 组合的情况下编写测试代码，此外它给出了一种在 observable 自身执行期间改变时间、并在特定点上做断言的方法。在本文中给出的所有这些技术，应该足够你选择用来测试 RxJava 的代码了。

未来

RxJava 是最先为 Java 提供响应式编程能力的程序库之一。为了使 RxJava API 更好地符合 Reactive Streams 规范，即将推出的 2.0 版将会是重新设计的。Reactive Streams 规范以 Java 和 JavaScript 运行时为目标，提供了使用非阻塞背压机制（back pressure）的异步流处理标准。这意味着下一版的 RxJava 中将会出现一些 API 改进。对这些改进的详细描述参见 RxJava wiki 。

对于测试而言，这些核心类型（Observable、Maybe 和 Single）现在都给出了便利易用的 test() 方法，实现现场创建 TestSubscriber 实例。也可在 TestSubscriber 上链接方法调用，对这类用法也有一些新的断言方法。

本文是“测试RxJava ”一文的修订。

关于作者

Andres Almiray是一位 Java/Groovy 开发者和 Java 冠军程序员，具有超过 17 年的软件设计和开发经验。他在 Java 早期推出的时代就参与了 Web 和桌面应用的开发。他是开源软件的忠实信徒，参与了 Groovy、JMatter、Asciiidoctor 等广为人知的项目。他是 Griffon 架构的创始者和现任项目领导者，还是 JSR 377 规范的牵头者。

查看英文原文： Testing RxJava2

感谢薛命灯对本文的审校。

给InfoQ 中文站投稿或者参与内容翻译工作，请邮件至 editors@cn.infoq.com 。也欢迎大家通过新浪微博（ @InfoQ ， @丁晓昀），微信（微信号： InfoQChina ）关注我们。

发布

暂无评论

促进软件开发及相关领域知识与创新的传播
关于我们
我要投稿
合作伙伴
加入我们
关注我们
联系我们
内容投稿：editors@geekbang.com
业务合作：hezuo@geekbang.com
反馈投诉：feedback@geekbang.com
加入我们：zhaopin@geekbang.com
联系电话：010-64738142
地址：北京市朝阳区望京北路9号2幢7层A701
InfoQ 近期会议
北京 · QCon全球软件开发大会 2024.4.18-20
北京 · AICon 全球人工智能与大模型应用峰会 2024.5.17-18
深圳 · ArchSummit全球架构师峰会 2024.6.14-15
上海 · FCon全球金融科技大会 2024.8.16-17
全球 InfoQ
InfoQ En
InfoQ Jp
InfoQ Fr
InfoQ Br

创作场景

测试 RxJava2

关键要点：

第一步

定制调度器（Scheduler）

操控时间

未来

关于作者

评论

云小课｜RDS for MySQL参数模板一键导入导出，参数配置轻松搞定

云计算架构设计原则

WeOpsV4.1重磅出击，日志监控功能横空出世！

中企出海之宝岛台湾假勤管理

GOTC峰会Sermant发布1.1.0-beta版本，带来哪些提升？

开源教育决定未来｜2023开放原子全球开源峰会开源教育与人才分论坛成功召开

人体分割识别图像技术的挑战和未来发展

洞察开源代码平台“创新力”｜2023开放原子全球开源峰会开源协作平台分论坛成功举办

华为云CodeArtBuild减负！云端编译构建，让你的开发省时省力！

Spring框架中的线程池

vivo 帐号服务稳定性建设之路-平台产品系列06

基于Spark的大规模日志分析

C语言代码封装MQTT协议报文，了解MQTT协议通信过程

微软增加100亿美元AI收入，Salesforce如何对垒？

六大类型JavaScript题型

解析Spring内置作用域及其在实践中的应用

百度文心一格x京东618：打造电商行业首个AI线下广告，节省80%制作成本

把数字中国，建立在行业感知的底座上

TrafficRoute：一体化的DNS解析和流量调度套件

全球化财务助力跨国业务的稳定增长和持续发展

开战在即！与全球伙伴一起打造你的数据应用，TiDB Future App Hackathon 2023 来啦！

2023-06-15：说一说Redis的Key和Value的数据结构组织?

阿里云斩获 4 项年度云原生技术服务优秀案例

HarmonyOS 极客马拉松2023 正式启动，诚邀极客们用键盘码出无限可能！

互联网行业-镭速文件传输系统方案

Django笔记四十三之使用uWSGI部署Django系统

C语言编程语法—排序算法

当AI大模型遇见金融，这四大挑战不容忽视

你一定要悄悄学会怎么把Vimeo视频保存到手机相册，然后惊艳所有人！

人体分割识别图像技术的原理及应用

UI自动化的稳定性和效率

	import io.reactivex.Observable;
	import io.reactivex.observers.TestObserver;
	import io.reactivex.plugins.RxJavaPlugins;
	import io.reactivex.schedulers.Schedulers;
	//RxJava2 中，包名由 rx.xxx 改为 io.reactivex.xxx。
	import org.junit.Test;
	import java.util.*;
	import static java.util.concurrent.TimeUnit.SECONDS;
	import static org.awaitility.Awaitility.await;
	import static org.junit.Assert.assertThat;
	import static org.hamcrest.Matchers.*;


	public class RxJavaTest {
	private static final List<String> WORDS = Arrays.asList(
	"the",
	"quick",
	"brown",
	"fox",
	"jumped",
	"over",
	"the",
	"lazy",
	"dog"
	);
	}

	@Test
	public void testInSameThread() {
	// given:
	List<String> results = new ArrayList<>();
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	observable.subscribe(results::add);

	// then:
	assertThat(results, notNullValue());
	assertThat(results, hasSize(9));
	assertThat(results, hasItem(" 4. fox"));
	}

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testUsingTestObserver() {
	// given:
	TestObserver<String> observer = new TestObserver<>();
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	observable.subscribe(observer);

	// then:
	observer.assertComplete();
	observer.assertNoErrors();
	observer.assertValueCount(9);
	//getOnNextEvents 方法该为 values 方法。
	assertThat(observer.values(), hasItem(" 4. fox"));
	}

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testFailure() {
	// given:
	TestObserver<String> observer = new TestObserver<>();
	Exception exception = new RuntimeException("boom!");

	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string))
	.concatWith(Observable.error(exception));

	// when:
	observable.subscribe(observer);

	// then:
	observer.assertError(exception);
	observer.assertNotComplete();
	}

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testUsingComputationScheduler() {
	// given:
	TestObserver<String> observer = new TestObserver<>();
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	observable.subscribeOn(Schedulers.computation())
	.subscribe(observer);

	// 修订版中新添加语句。
	await().timeout(2, SECONDS)
	.until(observer::valueCount, equalTo(9));

	// then:
	observer.assertComplete();
	observer.assertNoErrors();
	assertThat(observer.values(), hasItem(" 4. fox"));
	}

	@Test
	public void testUsingBlockingCall() {
	// given:
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	//RxJava2 中，toBlocking() 和 toIterable() 方法改为 blockingIterable()
	Iterable<String> results = observable
	.subscribeOn(Schedulers.computation())
	.blockingIterable();

	// then:
	assertThat(results, notNullValue());
	assertThat(results, iterableWithSize(9));
	assertThat(results, hasItem(" 4. fox"));
	}

	// 修订版中新添加的代码段。
	@Test
	public void testUsingComputationScheduler() {
	// given:
	TestObserver<String> observer = new TestObserver<>();
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	observable.subscribeOn(Schedulers.computation())
	.subscribe(observer);

	observer.awaitTerminalEvent(2, SECONDS);

	// then:
	observer.assertComplete();
	observer.assertNoErrors();
	assertThat(observer.values(), hasItem(" 4. fox"));
	}

	<dependency>
	<groupId>org.awaitility</groupId>
	<artifactId>awaitility</artifactId>
	<version>2.0.0</version>
	<scope>test</scope>
	</dependency>

	@Test
	// 名称中的 Hook 改为 Plugins，代码中所有 subscriber 改为 observer。
	public void testUsingRxJavaPluginsWithImmediateScheduler() {
	// given:
	// 调度器由 immediate 改为 trampoline。
	RxJavaPlugins.setComputationSchedulerHandler(scheduler ->
	Schedulers.trampoline());
	TestObserver<String> observer = new TestObserver<>();
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	try {
	// when:
	observable.subscribeOn(Schedulers.computation())
	.subscribe(observer);

	// then:
	observer.assertComplete();
	observer.assertNoErrors();
	observer.assertValueCount(9);
	assertThat(observer.values(), hasItem(" 4. fox"));
	} finally {
	RxJavaPlugins.reset();
	}
	}

	// 由 Public 改为 private static
	private static class ImmediateSchedulersRule implements TestRule {
	@Override
	public Statement apply(final Statement base, Description description) {
	return new Statement() {
	@Override
	// 所有的调度器由 immediate 改为 trampline。
	public void evaluate() throws Throwable {
	RxJavaPlugins.setIoSchedulerHandler(scheduler ->
	Schedulers.trampoline());
	RxJavaPlugins.setComputationSchedulerHandler(scheduler ->
	Schedulers.trampoline());
	RxJavaPlugins.setNewThreadSchedulerHandler(scheduler ->
	Schedulers.trampoline());

	try {
	base.evaluate();
	} finally {
	RxJavaPlugins.reset();
	}
	}
	};
	}
	}

	@Rule
	public final ImmediateSchedulersRule schedulers =
	new ImmediateSchedulersRule();

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testUsingImmediateSchedulersRule() {
	// given:
	TestObserver<String> observer = new TestObserver<>();
	//Observable 的 from 方法改为 fromIterable。
	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.range(1, Integer.MAX_VALUE),
	(string, index) -> String.format("%2d. %s", index, string));

	// when:
	observable.subscribeOn(Schedulers.computation())
	.subscribe(observer);

	// then:
	observer.assertComplete();
	observer.assertNoErrors();
	observer.assertValueCount(9);
	assertThat(observer.values(), hasItem(" 4. fox"));
	}

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testUsingTestScheduler() {
	// given:
	TestScheduler scheduler = new TestScheduler();
	TestObserver<String> observer = new TestObserver<>();
	Observable<Long> tick = Observable.interval(1, SECONDS, scheduler);

	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(tick,
	(string, index) -> String.format("%2d. %s", index, string));

	observable.subscribeOn(scheduler)
	.subscribe(observer);

	// expect:
	observer.assertNoValues();
	observer.assertNotComplete();

	// when:
	scheduler.advanceTimeBy(1, SECONDS);

	// then:
	observer.assertNoErrors();
	observer.assertValueCount(1);
	observer.assertValues(" 0. the");

	// when:
	scheduler.advanceTimeTo(9, SECONDS);
	observer.assertComplete();
	observer.assertNoErrors();
	observer.assertValueCount(9);
	}

	// 由 public 改为 private static，所有 Hooks 改为 Plugins。
	private static class TestSchedulerRule implements TestRule {
	private final TestScheduler testScheduler = new TestScheduler();

	public TestScheduler getTestScheduler() {
	return testScheduler;
	}

	@Override
	public Statement apply(final Statement base, Description description) {
	return new Statement() {
	@Override
	public void evaluate() throws Throwable {
	RxJavaPlugins.setIoSchedulerHandler(scheduler ->
	testScheduler);
	RxJavaPlugins.setComputationSchedulerHandler(scheduler ->
	testScheduler);
	RxJavaPlugins.setNewThreadSchedulerHandler(scheduler ->
	testScheduler);

	try {
	base.evaluate();
	} finally {
	RxJavaPlugins.reset();
	}
	}
	};
	}
	}

	@Rule
	public final TestSchedulerRule testSchedulerRule = new TestSchedulerRule();

	@Test
	// 例子中所有的 Subscriber 改为 Observer。
	public void testUsingTestSchedulersRule() {
	// given:
	TestObserver<String> observer = new TestObserver<>();

	Observable<String> observable = Observable.fromIterable(WORDS)
	.zipWith(Observable.interval(1, SECONDS),
	(string, index) -> String.format("%2d. %s", index, string));

	observable.subscribeOn(Schedulers.computation())
	.subscribe(observer);

	// expect
	observer.assertNoValues();
	observer.assertNotComplete();

	// when:
	testSchedulerRule.getTestScheduler().advanceTimeBy(1, SECONDS);

	// then:
	observer.assertNoErrors();
	observer.assertValueCount(1);
	observer.assertValues(" 0. the");

	// when:
	testSchedulerRule.getTestScheduler().advanceTimeTo(9, SECONDS);
	observer.assertComplete();
	observer.assertNoErrors();
	observer.assertValueCount(9);
	}

创作场景

测试 RxJava2

关键要点：

第一步

定制调度器（Scheduler）

操控时间

未来

关于作者

评论

更多内容推荐

推荐阅读

电子书

大厂实战PPT下载