Reactor Core实现了Reactive Streams规范，并处理了（可能无限的）数据流。 如果您感兴趣，请查看它提供的出色文档 。 在这里，我假设对Reactor Core库的Flux和Mono类型有一些基本的了解，并且将介绍Reactor Core提供了对时间本身的抽象，从而可以测试依赖于时间的函数。

对于某些Reactor核心运营商来说，时间是一个重要的考虑因素-例如，“间隔”功能的一种变体，它在初始“延迟” 10秒后每5秒发出一个递增的数字：

val flux = Flux.interval(Duration.ofSeconds(10), Duration.ofSeconds(5)).take(3)

根据正常时间流逝测试这样的数据流将是可怕的，这样的测试大约需要20秒才能完成。

Reactor-Core提供了一种解决方案，一种对时间本身的抽象-基于虚拟时间的调度程序，它提供了一种确定性的方式来测试这些类型的操作的巧妙方法。

让我以两种方式展示它，一种明确的方式应该使基于虚拟时间的调度程序的动作非常清晰，然后再推荐使用Reactor Core进行测试的方法。

import org.assertj.core.api.Assertions.assertThat
import org.junit.Test
import reactor.core.publisher.Flux
import reactor.test.scheduler.VirtualTimeScheduler
import java.time.Duration
import java.util.concurrent.CountDownLatchclass VirtualTimeTest {@Testfun testExplicit() {val mutableList = mutableListOf<Long>()val scheduler = VirtualTimeScheduler.getOrSet()val flux = Flux.interval(Duration.ofSeconds(10), Duration.ofSeconds(5), scheduler).take(3)val latch = CountDownLatch(1)flux.subscribe({ l -> mutableList.add(l) }, { _ -> }, { latch.countDown() })scheduler.advanceTimeBy(Duration.ofSeconds(10))assertThat(mutableList).containsExactly(0L)scheduler.advanceTimeBy(Duration.ofSeconds(5))assertThat(mutableList).containsExactly(0L, 1L)scheduler.advanceTimeBy(Duration.ofSeconds(5))assertThat(mutableList).containsExactly(0L, 1L, 2L)latch.await()}}

1.首先，将“ Flux.interval”功能的计划程序设置为基于虚拟时间的计划程序。

2.预计在10秒延迟后每5秒发射一次数据流

3. VirtualTimeScheduler提供了一种“ advanceTimeBy”方法来将虚拟时间提前一个持续时间，因此该时间将首先提前10秒的延迟时间，届时将发出第一个元素（0）。

4.然后将其前进5秒钟两次，分别得到1和2。

这是确定性的，测试可以快速完成。 但是，此版本的测试很丑陋，它使用列表来收集和声明结果，并使用CountDownLatch控制何时终止测试。 测试Reactor-Core类型的一种更为简洁的方法是使用出色的StepVerifier类，并且使用该类的测试如下所示：

import org.junit.Test
import reactor.core.publisher.Flux
import reactor.test.StepVerifier
import reactor.test.scheduler.VirtualTimeScheduler
import java.time.Durationclass VirtualTimeTest {@Testfun testWithStepVerifier() {VirtualTimeScheduler.getOrSet()val flux = Flux.interval(Duration.ofSeconds(10), Duration.ofSeconds(5)).take(3)StepVerifier.withVirtualTime({ flux }).expectSubscription().thenAwait(Duration.ofSeconds(10)).expectNext(0).thenAwait(Duration.ofSeconds(5)).expectNext(1).thenAwait(Duration.ofSeconds(5)).expectNext(2).verifyComplete()}}

借助StepVerifier进行的这项新测试可以很好地理解每步前进的时间，并断言当时的期望值。

翻译自: https://www.javacodegeeks.com/2017/09/testing-time-based-reactor-core-streams-virtual-time.html