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

2023-11-11 22:20
文章标签 java 技术 常见 检测 测试 rx 问题解答 停滞

本文主要是介绍技术大停滞_检测和测试停滞的流– RxJava常见问题解答，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

技术大停滞

技术大停滞

假设您有一个流以不可预测的频率发布事件。 有时您可以预期每秒会有数十条消息，但是偶尔几秒钟都看不到任何事件。 如果流是通过Web套接字，SSE或任何其他网络协议传输的，则可能会出现问题。 静默时间过长(停顿)可以解释为网络问题。 因此，我们经常不时发送人造事件( ping )，以确保：

  • 客户还活着
  • 让客户知道我们还活着

举一个更具体的例子，假设我们有一个Flowable<String>流，它会产生一些事件。 如果没有事件超过一秒钟，我们应该发送一个占位符"PING"消息。 当寂静时间更长时，应该每秒发出一个"PING"消息。 我们如何在RxJava中实现这样的要求？ 最明显但不正确的解决方案是将原始流与ping合并： 

Flowable<String> events = //...
Flowable<String> pings = Flowable.interval(1, SECONDS).map(x -> "PING");Flowable<String> eventsWithPings = events.mergeWith(pings);

mergeWith()运算符至关重要：它接受真正的events ，并将它们与恒定的ping流合并。 当然，当不存在真正的事件时，将显示"PING"消息。 不幸的是，它们与原始流完全无关。 这意味着即使有很多正常事件，我们也会继续发送ping命令。 而且，当静默开始时，我们不会在一秒钟后精确发送"PING" 。 如果您对这种机制感到满意，则可以在此处停止阅读。

一种更复杂的方法需要发现持续超过1秒的静音。 我们可以使用timeout()运算符。 不幸的是，它会产生TimeoutException并从上游退订-行为过于激进。 我们只想收到某种通知。 事实证明，可以使用debounce()运算符。 通常，此操作员会推迟新事件的发出，以防万一有新事件出现，而又覆盖了旧事件。 所以，如果我说： 

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);

这意味着delayed流仅在1秒内未跟随其他事件的情况下才会发出事件。 如果events流使产生事件的速度足够快，则从技术上讲， delayed可能永远不会发出任何东西。 我们将使用delayed流通过以下方式发现沉默： 

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.map(ev -> "PING");
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

请记住， mergeWith()与它的static merge()对应项之间没有区别。 所以我们到了某个地方。 如果流繁忙，则delayed流将永远不会收到任何事件，因此不会发送"PING"消息。 但是，当原始流不发送任何事件超过1秒时， delayed接收到最后一次看到的事件，将其忽略并转换为"PING" 。 聪明，但坏了。 此实现仅在发现停顿之后才发送一个"PING" ，而不是每秒发送一次定期ping。 很容易修复！ 除了将最后一次看到的事件转换为单个"PING"我们还可以将其转换为周期性的ping序列： 

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.flatMap(x -> Flowable.interval(0, 1, SECONDS).map(e -> "PING"));
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

您能看到缺陷在哪里吗？ 每当原始流中出现一点沉默时，我们就会每秒发出一次ping 。 但是，一旦出现真正的事件，我们应该停止这样做。 我们没有。 上游的每个停顿都会导致新的无限ping流出现在最终合并的流中。 我们必须以某种方式告诉pings流，因为原始流发出了真正的事件，所以它应该停止发出ping 。 猜猜是什么，有takeUntil()运算符可以做到这一点！ 

Flowable<String> events = //...
Flowable<String> delayed = events.debounce(1, SECONDS);
Flowable<String> pings = delayed.flatMap(x -> Flowable.interval(0, 1, SECONDS).map(e -> "PING").takeUntil(events));
Flowable<String> eventsWithPings = Flowable.merge(events, pings);

花一点时间完全掌握上面的代码片段。 每当原始流上超过1秒没有任何React时， delayed流就会发出一个事件。 pings流发射的序列"PING"每秒从发射每个事件的事件delayed 。 但是，一旦事件出现在events流上，便会终止pings流。 您甚至可以将所有这些定义为单个表达式： 

Flowable<String> events = //...
Flowable<String> eventsWithPings = events.mergeWith(events.debounce(1, SECONDS).flatMap(x1 -> Flowable.interval(0, 1, SECONDS).map(e -> "PING").takeUntil(events)));

可测性

好的，我们已经编写了所有这些内容，但是我们应该如何测试事件驱动代码的这个三层嵌套的Blob？ 我们如何确保ping在正确的时间出现并在静音结束后停止？ 如何模拟各种与时间相关的场景？ RxJava具有许多杀手级功能，但是测试时间流逝可能是最大的功能。 首先，让我们的ping代码更具可测试性和通用性： 

<T> Flowable<T> withPings(Flowable<T> 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 ，以防该流在较长时间内不产生任何事件。 我们在测试中像这样使用它： 

PublishProcessor<String> events = PublishProcessor.create();
TestScheduler clock = new TestScheduler();
Flowable<String> 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 {PublishProcessor<String> events = PublishProcessor.create();final TestScheduler clock = new TestScheduler();final Flowable<String> eventsWithPings = withPings(events, clock, "PING");final TestSubscriber<String> 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");
}

看起来像一堵墙，但这实际上是我们逻辑的完整测试方案。 它可以确保在1000毫秒后准确地执行ping操作；如果寂静时间很长，则会重复进行ping操作；当出现真正的事件时， ping操作会变得很慢。 但最重要的部分是：该测试是100％可预测的并且非常快。 没有Awaitility ，忙等待，轮询，间歇性测试失败和缓慢。 我们完全控制的人工时钟可确保所有这些组合流均按预期工作。

翻译自: https://www.javacodegeeks.com/2017/09/detecting-testing-stalled-streams-rxjava-faq.html

技术大停滞

这篇关于技术大停滞_检测和测试停滞的流– RxJava常见问题解答的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/393213。

相关文章

JVM 的类初始化机制

JVM 的类初始化机制

前言 当你在 Java 程序中new对象时，有没有考虑过 JVM 是如何把静态的字节码（byte code）转化为运行时对象的呢，这个问题看似简单，但清楚的同学相信也不会太多，这篇文章首先介绍 JVM 类初始化的机制，然后给出几个易出错的实例来分析，帮助大家更好理解这个知识点。 JVM 将字节码转化为运行时对象分为三个阶段，分别是：loading 、Linking、initialization
阅读更多...
Spring Security 基于表达式的权限控制

Spring Security 基于表达式的权限控制

前言 spring security 3.0已经可以使用spring el表达式来控制授权，允许在表达式中使用复杂的布尔逻辑来控制访问的权限。 常见的表达式 Spring Security可用表达式对象的基类是SecurityExpressionRoot。 表达式描述hasRole([role])用户拥有制定的角色时返回true （Spring security默认会带有ROLE_前缀）,去
阅读更多...
浅析Spring Security认证过程

浅析Spring Security认证过程

类图 为了方便理解Spring Security认证流程，特意画了如下的类图，包含相关的核心认证类 概述 核心验证器 AuthenticationManager 该对象提供了认证方法的入口，接收一个Authentiaton对象作为参数; public interface AuthenticationManager {Authentication authenticate(Authenti
阅读更多...
Spring Security--Architecture Overview

Spring Security--Architecture Overview

1 核心组件 这一节主要介绍一些在Spring Security中常见且核心的Java类，它们之间的依赖，构建起了整个框架。想要理解整个架构，最起码得对这些类眼熟。 1.1 SecurityContextHolder SecurityContextHolder用于存储安全上下文（security context）的信息。当前操作的用户是谁，该用户是否已经被认证，他拥有哪些角色权限…这些都被保
阅读更多...
Spring Security基于数据库验证流程详解

Spring Security基于数据库验证流程详解

Spring Security 校验流程图 相关解释说明（认真看哦） AbstractAuthenticationProcessingFilter 抽象类 /*** 调用 #requiresAuthentication(HttpServletRequest, HttpServletResponse) 决定是否需要进行验证操作。* 如果需要验证，则会调用 #attemptAuthentica
阅读更多...
Spring Security 从入门到进阶系列教程

Spring Security 从入门到进阶系列教程

Spring Security 入门系列 《保护 Web 应用的安全》 《Spring-Security-入门（一）：登录与退出》 《Spring-Security-入门（二）：基于数据库验证》 《Spring-Security-入门（三）：密码加密》 《Spring-Security-入门（四）：自定义-Filter》 《Spring-Security-入门（五）：在 Sprin
阅读更多...
Java架构师知识体认识

Java架构师知识体认识

源码分析 常用设计模式 Proxy代理模式Factory工厂模式Singleton单例模式Delegate委派模式Strategy策略模式Prototype原型模式Template模板模式 Spring5 beans 接口实例化代理Bean操作 Context Ioc容器设计原理及高级特性Aop设计原理Factorybean与Beanfactory Transaction 声明式事物
阅读更多...
性能测试介绍

性能测试介绍

性能测试是一种测试方法，旨在评估系统、应用程序或组件在现实场景中的性能表现和可靠性。它通常用于衡量系统在不同负载条件下的响应时间、吞吐量、资源利用率、稳定性和可扩展性等关键指标。 为什么要进行性能测试 通过性能测试，可以确定系统是否能够满足预期的性能要求，找出性能瓶颈和潜在的问题，并进行优化和调整。 发现性能瓶颈：性能测试可以帮助发现系统的性能瓶颈，即系统在高负载或高并发情况下可能出现的问题
阅读更多...
Java进阶13讲__第12讲_1/2

Java进阶13讲__第12讲_1/2

多线程、线程池 1.  线程概念 1.1  什么是线程 1.2  线程的好处 2.   创建线程的三种方式 注意事项 2.1  继承Thread类 2.1.1 认识  2.1.2  编码实现  package cn.hdc.oop10.Thread;import org.slf4j.Logger;import org.slf4j.LoggerFactory
阅读更多...
字节面试 | 如何测试RocketMQ、RocketMQ？

字节面试 | 如何测试RocketMQ、RocketMQ？

字节面试：RocketMQ是怎么测试的呢？ 答： 首先保证消息的消费正确、设计逆向用例，在验证消息内容为空等情况时的消费正确性； 推送大批量MQ，通过Admin控制台查看MQ消费的情况，是否出现消费假死、TPS是否正常等等问题。（上述都是临场发挥，但是RocketMQ真正的测试点，还真的需要探讨） 01 先了解RocketMQ 作为测试也是要简单了解RocketMQ。简单来说，就是一个分
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2