固定速率与固定延迟– RxJava常见问题解答

如果您使用的是纯Java，从版本5开始，我们有一个方便的调度程序类，该类允许以固定速率或固定延迟运行任务：

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(10);

基本上，它支持两种类型的操作：

scheduler.scheduleAtFixedRate(() -> doStuff(), 2, 1, SECONDS);
scheduler.scheduleWithFixedDelay(() -> doStuff(), 2, 1, SECONDS);

scheduleAtFixedRate()将确保每秒精确调用doStuff()且初始延迟为2秒。当然，垃圾回收，上下文切换等仍然会影响精度。 scheduleWithFixedDelay()看起来很相似，但是它考虑了doStuff()处理时间。例如，如果doStuff()运行doStuff()毫秒，则固定速率将仅等待800毫秒，直到下一次重试。另一方面， scheduleWithFixedDelay()总是在重试之间等待相同的时间（在本例中为1秒）。在不同情况下，两种行为当然都是可取的。仅记住，当doStuff()的速度慢于1秒时， scheduleAtFixedRate()不会保留所需的频率。即使我们的ScheduledExecutorService有10个线程， doStuff()也绝不会被同时调用并且不会与之前的执行重叠。因此，在这种情况下，速率实际上将小于配置的速率。
<h1”> RxJava中的计划

使用interval()运算符，使用RxJava模拟scheduleAtFixedRate()非常简单。有几点警告：

Flowable.interval(2, 1, SECONDS).subscribe(i -> doStuff());

如果doStuff()的速度慢于1秒，则会发生不良情况。首先，我们使用Schedulers.computation()线程池，它是从interval()运算符继承的默认池。这是一个坏主意，该线程池仅应用于CPU密集型任务，并在整个RxJava之间共享。一个更好的主意是使用您自己的调度程序（或至少使用io() ）：

Flowable.interval(2, 1, SECONDS).observeOn(Schedulers.io()).subscribe(i -> doStuff());

observeOn()从开关computation()由用于调度interval()到io()调度器。由于subscribe()方法永远不会被设计并发调用， doStuff()永远不会并发调用，就像scheduleAtFixedRate() 但是， interval()运算符非常努力地保持恒定的频率。这意味着如果过一会儿doStuff()的速度慢于1秒，我们应该期望MissingBackpressureException …RxJava基本上告诉我们订户速度太慢，但是interval() （根据设计）不会变慢。如果您允许（甚至期望） doStuff()并发执行重叠，则修复起来非常简单。首先，您必须使用非阻塞式Completable包装阻塞式doStuff() 。从技术上讲， Flowable Single或Maybe也可以工作，但是由于doStuff()为void ，所以Completable听起来不错：

import io.reactivex.Completable;
import io.reactivex.schedulers.Schedulers;Completable doStuffAsync() {return Completable.fromRunnable(this::doStuff).subscribeOn(Schedulers.io()).doOnError(e -> log.error("Stuff failed", e)).onErrorComplete();
}

捕获并吞下异常很重要，否则单个错误将导致整个interval()中断。 doOnError()允许记录日志，但它通过下游传递异常。另一方面， doOnComplete()仅吞下异常。现在，我们可以在每个间隔事件中简单地运行此操作

Flowable.interval(2, 1, SECONDS).flatMapCompletable(i -> doStuffAsync()).subscribe();

如果您不subscribe()循环将永远不会开始-但这是RxJava101。请注意，如果doStuffAsync()花费一秒钟以上的时间来完成，我们将得到重叠的并发执行。这没有什么错，您只需要意识到这一点。但是，如果您真正需要的是固定延迟怎么办？

修复了RxJava中的延迟

在某些情况下，您需要固定的延迟时间：任务不应重叠，并且我们应在两次执行之间保持一定的喘息时间。不管周期性任务有多慢，都应始终保持恒定的时间暂停。 interval()运算符不适合实现此要求。但是，事实证明，RxJava中的解决方案非常简单。想一想：您需要睡一会儿，运行一些任务，然后在完成此任务后重复。让我再说一遍：

睡一会儿（有一些timer() ）
运行一些任务，等待它complete()
repeat()

而已！

Flowable.timer(1, SECONDS).flatMapCompletable(i -> doStuffAsync()).repeat().subscribe();

一秒钟后， timer()运算符发出一个事件（ Long类型的0 ）。我们使用此事件来触发doStuffAsync() 。当我们的东西做，全码流完成-但我们想重复！好吧， repeat()运算符就是这样做的：当它从上游收到完成通知时，它会重新订阅。重新订阅基本上意味着：再等待1秒钟， doStuffAsync() –依此类推。