上海营销型网站建设,wordpress根据字段判断,lnmp WordPress 改端口,html购物网站源码技术停滞假设您有一个流以不可预测的频率发布事件。 有时您可以预期每秒会有数十条消息#xff0c;但是偶尔几秒钟都看不到任何事件。 如果您的流是通过Web套接字#xff0c;SSE或任何其他网络协议传输的#xff0c;则可能会出现问题。 静默时间过长#xff08;停顿#x… 技术停滞 假设您有一个流以不可预测的频率发布事件。 有时您可以预期每秒会有数十条消息但是偶尔几秒钟都看不到任何事件。 如果您的流是通过Web套接字SSE或任何其他网络协议传输的则可能会出现问题。 静默时间过长停顿可以解释为网络问题。 因此我们经常不时发送人造事件 ping 以确保 客户还活着 让客户知道我们还活着 举一个更具体的例子假设我们有一个FlowableString流它会产生一些事件。 如果没有事件超过一秒钟我们应该发送一个占位符PING消息。 当静默时间更长时应该每秒发出一个PING消息。 我们如何在RxJava中实现这样的要求 最明显但不正确的解决方案是将原始流与ping合并 FlowableString events //...
FlowableString pings Flowable.interval(1, SECONDS).map(x - PING);FlowableString eventsWithPings events.mergeWith(pings); mergeWith()运算符至关重要它接受真正的events 并将它们与恒定的ping流合并。 当然当没有真实的事件出现时将出现PING消息。 不幸的是它们与原始流完全无关。 这意味着即使有很多正常事件我们也会继续发送ping命令。 而且当静默开始时我们不会在一秒钟后精确发送PING 。 如果您对这种机制感到满意则可以在此处停止阅读。 一种更复杂的方法需要发现持续超过1秒的静音。 我们可以使用timeout()运算符。 不幸的是它会产生TimeoutException并从上游退订-行为过于激进。 我们只想收到某种通知。 事实证明可以使用debounce()运算符。 通常此操作员会推迟新事件的发出以防万一有新事件出现从而覆盖了旧事件。 所以如果我说 FlowableString events //...
FlowableString delayed events.debounce(1, SECONDS); 这意味着delayed流仅在1秒内未跟随其他事件时才会发出事件。 如果events流使产生事件的速度足够快从技术上讲 delayed可能永远不会发出任何东西。 我们将使用delayed流通过以下方式发现沉默 FlowableString events //...
FlowableString delayed events.debounce(1, SECONDS);
FlowableString pings delayed.map(ev - PING);
FlowableString eventsWithPings Flowable.merge(events, pings); 请记住 mergeWith()和它的static merge()对应项之间没有区别。 所以我们到了某个地方。 如果流繁忙则delayed流将永远不会收到任何事件因此不会发送PING消息。 但是当原始流不发送任何事件超过1秒时 delayed接收到最后看到的事件将其忽略并转换为PING 。 聪明但坏了。 此实现仅在发现停顿后才发送一个PING 而不是每秒发送一次定期ping。 很容易修复 除了将最后一次看到的事件转换为单个PING我们还可以将其转换为周期性ping序列 FlowableString events //...
FlowableString delayed events.debounce(1, SECONDS);
FlowableString pings delayed.flatMap(x - Flowable.interval(0, 1, SECONDS).map(e - PING));
FlowableString eventsWithPings Flowable.merge(events, pings); 您能看到缺陷在哪里吗 每当原始流中出现一点沉默时我们就会每秒发出一次ping 。 但是一旦出现一些真正的事件我们应该停止这样做。 我们没有。 上游的每个停顿都会导致新的无限ping流出现在最终的合并流中。 我们必须以某种方式告诉pings流因为原始流发出了真正的事件所以它应该停止发出ping 。 猜猜是什么有takeUntil()运算符可以做到这一点 FlowableString events //...
FlowableString delayed events.debounce(1, SECONDS);
FlowableString pings delayed.flatMap(x - Flowable.interval(0, 1, SECONDS).map(e - PING).takeUntil(events));
FlowableString eventsWithPings Flowable.merge(events, pings); 花一点时间完全掌握上面的代码片段。 每当原始流上超过1秒没有任何React时 delayed流就会发出一个事件。 pings流发射的序列PING每秒从发射每个事件的事件delayed 。 但是一旦事件出现在events流上便会终止pings流。 您甚至可以将所有这些定义为单个表达式 FlowableString events //...
FlowableString eventsWithPings events.mergeWith(events.debounce(1, SECONDS).flatMap(x1 - Flowable.interval(0, 1, SECONDS).map(e - PING).takeUntil(events)));可测性 好的我们已经编写了所有这些内容但是我们应该如何测试事件驱动代码的这个三层嵌套的Blob 我们如何确保ping在正确的时间出现并在静默结束时停止 如何模拟各种与时间相关的场景 RxJava具有许多杀手级功能但是测试时间流逝可能是最大的功能。 首先让我们的ping代码更具可测试性和通用性 T FlowableT withPings(FlowableT events, Scheduler clock, T ping) {return events.mergeWith(events.debounce(1, SECONDS, clock).flatMap(x1 - Flowable.interval(0, 1, SECONDS, clock).map(e - ping).takeUntil(events)));} 此实用程序方法采用任意的T流并添加ping 以防该流在较长时间内不产生任何事件。 我们在测试中像这样使用它 PublishProcessorString events PublishProcessor.create();
TestScheduler clock new TestScheduler();
FlowableString eventsWithPings withPings(events, clock, PING); 哦男孩 PublishProcessor TestScheduler PublishProcessor是一个有趣的类它是一个亚型Flowable 所以我们可以使用它作为一个普通的流。 另一方面我们可以使用其onNext()方法强制发出事件 events.onNext(A); 如果有人收听events流他将立即收到A事件。 这clock是怎么回事 RxJava中以任何方式处理时间的每个运算符例如debounce debounce() interval() timeout() window() 都可以采用可选的Scheduler参数。 它充当时间的外部来源。 特殊的TestScheduler是我们完全控制的人为的时间来源。 也就是说只要我们不显式调用advanceTimeBy()时间就保持静止 clock.advanceTimeBy(999, MILLISECONDS); 999毫秒不是巧合。 Ping在1秒钟后开始精确显示因此在999毫秒后将不可见。 现在是时候揭示完整的测试用例了 Test
public void shouldAddPings() throws Exception {PublishProcessorString events PublishProcessor.create();final TestScheduler clock new TestScheduler();final FlowableString eventsWithPings withPings(events, clock, PING);final TestSubscriberString test eventsWithPings.test();events.onNext(A);test.assertValues(A);clock.advanceTimeBy(999, MILLISECONDS);events.onNext(B);test.assertValues(A, B);clock.advanceTimeBy(999, MILLISECONDS);test.assertValues(A, B);clock.advanceTimeBy(1, MILLISECONDS);test.assertValues(A, B, PING);clock.advanceTimeBy(999, MILLISECONDS);test.assertValues(A, B, PING);events.onNext(C);test.assertValues(A, B, PING, C);clock.advanceTimeBy(1000, MILLISECONDS);test.assertValues(A, B, PING, C, PING);clock.advanceTimeBy(999, MILLISECONDS);test.assertValues(A, B, PING, C, PING);clock.advanceTimeBy(1, MILLISECONDS);test.assertValues(A, B, PING, C, PING, PING);clock.advanceTimeBy(999, MILLISECONDS);test.assertValues(A, B, PING, C, PING, PING);events.onNext(D);test.assertValues(A, B, PING, C, PING, PING, D);clock.advanceTimeBy(999, MILLISECONDS);events.onNext(E);test.assertValues(A, B, PING, C, PING, PING, D, E);clock.advanceTimeBy(999, MILLISECONDS);test.assertValues(A, B, PING, C, PING, PING, D, E);clock.advanceTimeBy(1, MILLISECONDS);test.assertValues(A, B, PING, C, PING, PING, D, E, PING);clock.advanceTimeBy(3_000, MILLISECONDS);test.assertValues(A, B, PING, C, PING, PING, D, E, PING, PING, PING, PING);
} 看起来像一堵墙但这实际上是我们逻辑的完整测试方案。 它可以确保ping在1000毫秒后精确显示当沉默很长时重复一次而在真正事件出现时则重复很短。 但最重要的部分是该测试是100可预测的并且非常快。 没有Awaitility 忙等待轮询间歇性测试失败和缓慢。 我们完全控制的人工时钟可确保所有这些组合流均按预期工作。 翻译自: https://www.javacodegeeks.com/2017/09/detecting-testing-stalled-streams-rxjava-faq.html技术停滞
