手机网站设置方法,网站栏目模板如何选择,wordpress设置首页标题描述,工程建设专业RxJava缺少创建无限自然数流的工厂。 这样的流很有用#xff0c;例如#xff0c;当您想通过压缩两个事件的唯一序列号给可能的无限事件流时#xff1a; FlowableLong naturalNumbers //???FlowableEvent someInfiniteEventStream //...
FlowableP… RxJava缺少创建无限自然数流的工厂。 这样的流很有用例如当您想通过压缩两个事件的唯一序列号给可能的无限事件流时 FlowableLong naturalNumbers //???FlowableEvent someInfiniteEventStream //...
FlowablePairLong, Event sequenced Flowable.zip(naturalNumbers,someInfiniteEventStream,Pair::of
); 实现naturalNumbers令人惊讶地复杂。 在RxJava 1.x中您可以短暂地放弃不遵守背压的Observable import rx.Observable; //RxJava 1.xObservableLong naturalNumbers Observable.create(subscriber - {long state 0;//poor solution :-(while (!subscriber.isUnsubscribed()) {subscriber.onNext(state);}
}); 这样的流没有背压是什么意思 好吧基本上流可以轻松地以CPU内核允许的速度生成事件 state变量不断增加每秒数百万。 但是当使用者无法快速使用事件时未处理事件的积压开始出现 naturalNumbers
// .observeOn(Schedulers.io()).subscribe(x - {//slooow, 1 millisecond}); 上面的程序带有observeOn()运算符的注释掉可以正常运行因为它具有意外的反压。 默认情况下所有内容在RxJava中都是单线程的因此生产者和使用者在同一个线程中工作。 实际上调用subscriber.onNext()会阻止因此while循环会自动对其进行限制。 但是尝试取消注释observeOn() 灾难会在几毫秒后发生。 订阅回调在设计上是单线程的。 对于每个元素它至少需要1毫秒因此该流每秒可以处理不超过1000个事件。 我们有些幸运。 RxJavaSwift发现这种灾难性状况并因MissingBackpressureException而快速失败 我们最大的错误是在生成事件时没有考虑消费者的速度。 顺便说一下这是响应流背后的核心思想不允许生产者发出比消费者请求更多的事件。 在RxJava 1.x中即使实现最简单的流从头开始考虑背压也不是一件容易的事。 RxJava 2.x带来了一些便利的运算符这些运算符建立在先前版本的经验基础之上。 首先RxJava 2.x时不允许你实现Flowable 背压-aware的相同的方式你可以与Observable 。 创建Flowable会使消费者使消息过载是不可能的 FlowableLong naturalNumbers Flowable.create(subscriber - {long state 0;while (!subscriber.isCancelled()) {subscriber.onNext(state);}
}, BackpressureStrategy.DROP); 您是否发现了这个额外的DROP参数 在解释之前让我们看一下使用慢速用户订阅时的输出 0
1
2
3
//...continuous numbers...
126
127
101811682
//...where did my 100M events go?!?
101811683
101811684
101811685
//...continuous numbers...
101811776
//...17M events disappeared again...
101811777
//... 你的旅费可能会改变。 怎么了 observeOn()运算符在调度程序线程池之间切换。 从未决事件队列中合并的线程池。 此队列是有限的容量为128个元素。 知道此限制的observeOn()运算符仅从上游请求128个元素我们的自定义Flowable 。 此时它使我们的订户可以处理事件每毫秒1次。 因此大约100毫秒后 observeOn()发现其内部队列几乎为空并要求更多。 会得到128、129、130…吗 没有 我们的Flowable在这0.1秒内产生了疯狂的事件并且令人惊讶地在该时间范围内成功产生了超过1亿个数字。 他们去哪了 好吧 observeOn()并没有要求它们因此DROP策略强制性参数只是丢弃了不需要的事件。 BackpressureStrategy 听起来不对还有其他策略吗 是的很多 BackpressureStrategy.BUFFER 如果上游产生太多事件则将它们缓冲在无界队列中。 没有任何事件丢失但是您的整个应用程序很可能会丢失。 如果幸运的话 OutOfMemoryError将拯救您。 我停留在5秒以上的GC暂停中。 BackpressureStrategy.ERROR 如果发现事件的过度产生将抛出MissingBackpressureException 。 这是一个理智安全的策略。 BackpressureStrategy.LATEST 类似于DROP 但是记住上次删除的事件。 万一要求提供更多数据但我们只是丢弃了所有内容–至少具有最后看到的价值。 BackpressureStrategy.MISSING 没有安全措施请加以处理。 下游运算符之一如observeOn() 最有可能抛出MissingBackpressureException 。 BackpressureStrategy.DROP 删除未请求的事件。 顺便说一句当您将Observable变为Flowable还必须提供BackpressureStrategy 。 RxJava必须知道如何限制过量产生的Observable 。 好的那么简单的序列自然数流的正确实现是什么 认识 create()和generate()之间的区别在于责任。 假设Flowable.create()会在不考虑背压的情况下完整地生成流。 它只是在需要时才产生事件。 另一方面仅允许Flowable.generate()一次生成一个事件或完成流。 背压机制透明地计算出当前需要多少个事件。 generate()调用适当的次数例如在observeOn()情况下 observeOn() 128次。 因为此运算符一次生成一个事件所以通常需要某种状态来确定上次出现的时间1 。 这就是generate()含义im可变状态的持有者和基于该状态生成下一个事件的函数 FlowableLong naturalNumbers Flowable.generate(() - 0L, (state, emitter) - {emitter.onNext(state);return state 1;}); generate()的第一个参数是初始状态工厂在本例中为0L 。 现在每当订户或任何下游操作员要求一定数量的事件时都会调用lambda表达式。 它的职责是根据提供的状态最多调用一次onNext() 最多发出一个事件。 首次调用lambda时 state等于初始值0L 。 但是我们可以修改状态并返回其新值。 在此示例中我们增加了long以便随后的lambda表达式调用收到state 1L 。 显然这种情况不断发生产生连续的自然数。 这样的编程模型显然比while循环难。 它还从根本上改变了实现事件源的方式。 与其在任何时候都想推送事件不如只是被动地等待请求。 下游运营商和订户正在从您的流中提取数据。 这种转变可在管道的所有级别上产生背压。 generate()有一些风格。 首先如果您的状态是可变对象则可以使用不需要返回新状态值的重载版本。 尽管功能较少但可变状态往往会产生较少的垃圾。 这假设您的状态不断变化并且每次都传递相同的状态对象实例。 例如您可以轻松地将Iterator 也是基于pull的变成具有反压奇观的流 IteratorInteger iter //...FlowableString strings Flowable.generate(() - iter, (iterator, emitter) - {if (iterator.hasNext()) {emitter.onNext(iterator.next().toString());} else {emitter.onComplete();}
}); 注意流的类型 String 不必与状态类型 IteratorInteger 相同。 当然如果您有Java Collection并想将其转换为流则不必先创建迭代器。 使用Flowable.fromIterable()足够了。 甚至更简单的generate()版本都假定您根本没有任何状态。 例如随机数流 FlowableDouble randoms Flowable.generate(emitter - emitter.onNext(Math.random())); 但老实说您可能最终将需要一个Random实例 Flowable.generate(Random::new, (random, emitter) - {emitter.onNext(random.nextBoolean());
});摘要 如您所见RxJava 1.x中的Observable.create()和Flowable.create Flowable.create()有一些缺点。 如果您真的在乎大量并发系统的可伸缩性和运行状况否则您将不会读到这篇文章则必须了解背压。 如果您真的需要从头开始创建流而不是使用from*()系列方法或繁重工作的各种库请熟悉generate() 。 本质上您必须学习如何将某些类型的数据源建模为奇特的迭代器。 期待有更多文章解释如何实现更多现实生活流。 这类似于无状态HTTP协议该协议在服务器上使用称为session *的小状态来跟踪过去的请求。 翻译自: https://www.javacodegeeks.com/2017/08/generating-backpressure-aware-streams-flowable-generate-rxjava-faq.html
