如果您使用的是纯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 SingleMaybe也可以工作,但是由于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() –依此类推。

翻译自: https://www.javacodegeeks.com/2017/09/fixed-rate-vs-fixed-delay-rxjava-faq.html

固定速率与固定延迟– RxJava常见问题解答相关推荐

  1. openwrt固定速率_固定速率与固定延迟– RxJava常见问题解答

    openwrt固定速率 如果您使用的是纯Java,从版本5开始,我们有一个方便的调度程序类,该类允许以固定速率或固定延迟运行任务: import java.util.concurrent.Execut ...

  2. 渴望 英语_渴望订阅– RxJava常见问题解答

    渴望 英语 在教学和指导RxJava以及撰写本书之后 ,我注意到某些领域特别有问题. 我决定发布一些简短的提示,以解决最常见的陷阱. 这是第一部分. Observable和Flowable本质上是惰性 ...

  3. 渴望订阅– RxJava常见问题解答

    在教学和指导RxJava以及撰写本书之后 ,我注意到某些领域尤其成问题. 我决定发布一些简短的提示,以解决最常见的陷阱. 这是第一部分. Observable和Flowable本质上是惰性的. 这意味 ...

  4. flatmap_flatMap()与concatMap()与concatMapEager()– RxJava常见问题解答

    flatmap RxJava 2.x中共有三个无缝相似的运算符: flatMap() , concatMap()和concatMapEager() . 它们都接受相同的参数-从原始流的单个项目到任意类 ...

  5. flatMap()与concatMap()与concatMapEager()– RxJava常见问题解答

    RxJava 2.x中共有三个无缝相似的运算符: flatMap() , concatMap()和concatMapEager() . 它们都接受相同的参数-从原始流的单个项目到任意类型的(子)流的函 ...

  6. 技术停滞_检测和测试停滞的流– RxJava常见问题解答

    技术停滞 假设您有一个流以不可预测的频率发布事件. 有时您可以预期每秒会有数十条消息,但是偶尔几秒钟都看不到任何事件. 如果您的流是通过Web套接字,SSE或任何其他网络协议传输的,则可能会出现问题. ...

  7. 迈克尔 杰克逊mv_用杰克逊流式传输大型JSON文件– RxJava常见问题解答

    迈克尔 杰克逊mv 在上一篇文章中,我们学习了如何解析过大的XML文件并将其转换为RxJava流. 这次让我们看一个大的JSON文件. 我们的示例将基于微小的colors.json,其中包含将近150 ...

  8. 并发查询parallel_惯用并发:flatMap()与parallel()– RxJava常见问题解答

    并发查询parallel 简单,有效和安全的并发是RxJava的设计原则之一. 然而,具有讽刺意味的是,它可能是该库中最容易被误解的方面之一. 让我们举一个简单的例子:假设我们有一堆UUID并且对于每 ...

  9. flowable背压 取消_使用Flowable.generate()生成可感知背压的流– RxJava常见问题解答...

    flowable背压 取消 RxJava缺少创建无限自然数流的工厂. 这样的流很有用,例如,当您想通过压缩两个事件的顺序来为可能的无限事件流分配唯一的序列号时: Flowable<Long> ...

最新文章

  1. parseIntRight
  2. 【Linux】一步一步学Linux——ssh命令(176)
  3. Docker Installation : Docker 中安装并启动 Kong
  4. python 截取字符串6位_在Python中从字符串获取x个最低有效位
  5. 【病毒】开机弹出“tlntsvi_6635.exe程序”解决方案
  6. 图层上下_「只要功夫深 不用关键帧」之图层序号的玩法
  7. OpenCV学习笔记(一):读取、显示、保存图片
  8. php学习的一些笔记
  9. python合并大量ts文件_python之爬虫(爬取.ts文件并将其合并为.MP4文件——以及一些异常的注意事项)...
  10. 匹配滤波器的MATLAB实现
  11. 联想Y450电源管理无法调节屏幕亮度的解决办法
  12. LaunchAnywhere载入Java VM时windows出现错误:2以及216
  13. 自学单片机入门学习建议
  14. Windows本地认证
  15. 计算机如何分割硬盘,电脑硬盘怎么分区 电脑硬盘分区方法大全
  16. Qt Creator 8.0.1 (Community)创建项目出现No valid kits found怎么办?
  17. 什么是拼多多通用券|盛天海科技
  18. 『居善地』接口测试 — 11、接口自动化框架的设计与实现(三)
  19. GitHub的raw.githubusercontent.com无法链接
  20. 汉诺塔问题的解决思路及算法

热门文章

  1. 搞定所有的跨域请求问题
  2. 10 道关于 Java 泛型的面试题
  3. MySQL dayname()函数
  4. MySQL find_in_set()函数
  5. JVM内存管理------垃圾搜集器参数精解
  6. 京东购物车的 Java 架构实现及原理
  7. 阿里巴巴对Java编程【控制语句】的规约
  8. avue中实现消息的实时展示
  9. java实现人脸识别(使用百度云V3版本)
  10. 利用老毛头启动盘重装win7