Mono 超时时的“运算符调用默认 onErrorDropped”

Question

在我的生产代码中，当 Mono 超时时，我的日志中出现错误。
我已设法使用以下代码重新创建这些错误：

@Test
public void testScheduler() {
    Mono<String> callableMethod1 = callableMethod();
    callableMethod1.block();

    Mono<String> callableMethod2 = callableMethod();
    callableMethod2.block();
}

private Mono<String> callableMethod() {
    return Mono.fromCallable(() -> {
        Thread.sleep(60);
        return "Success";
    })
            .subscribeOn(Schedulers.elastic())
            .timeout(Duration.ofMillis(50))
            .onErrorResume(throwable -> Mono.just("Timeout"));
}

在Mono.fromCallable我正在使用第三方库进行阻塞调用。当此调用超时时，我收到类似于

reactor.core.publisher.Operators - Operator called default onErrorDropped
reactor.core.publisher.Operators - Scheduler worker in group main failed with an uncaught exception

这些错误似乎也是间歇性的，有时当我运行代码时，我根本没有得到任何错误。但是，当我以 10 的循环重复调用时，我始终得到它们。

Answer 1

问题：为什么会发生这个错误？

回答：

当给timeout()运算符的持续时间已过时，它会抛出TimeoutException。这导致以下结果：

1. 一个onError信号被发送到主反应链。结果，主执行重新开始，进程继续进行（即执行onErrorResume()）。

2. 在结果 #1 之后不久，在fromCallable() 中定义的异步任务被中断，这会触发第二个异常 ( InterruptedException )。主反应链不能再处理这个InterruptedException因为TimeoutException首先发生并且已经导致主反应链恢复（注意：不生成第二个onError信号的行为符合反应流规范 -> Publisher #7）。

由于主链无法优雅地处理第二个异常（InterruptedException），Reactor 将其记录在错误级别，让我们知道发生了意外异常。

问题：我如何摆脱它们？

简短回答：使用Hooks.onErrorDropped()更改日志级别：

    Logger logger = Logger.getLogger(this.getClass().getName());
    @Test
    public void test() {
        Hooks.onErrorDropped(error -> {
            logger.log(Level.WARNING, "Exception happened:", error);
        });

        Mono.fromCallable(() -> {
            Thread.sleep(60);
            return "Success";
        })
         .subscribeOn(Schedulers.elastic())
         .timeout(Duration.ofMillis(50))
         .onErrorResume(throwable -> Mono.just("Timeout"))
         .doOnSuccess(result -> logger.info("Result: " + result))
         .block();
     }

长答案：如果您的用例允许，您可以处理fromCallable() 中发生的异常，以便影响主链的唯一异常是TimeoutException。在这种情况下，onErrorDropped() 一开始就不会发生。

@Test
public void test() {
    Mono.fromCallable(() -> {
        try {
            Thread.sleep(60);
        } catch (InterruptedException ex) {
            //release resources, rollback actions, etc
            logger.log(Level.WARNING, "Something went wrong...", ex);
        }
        return "Success";
    })
     .subscribeOn(Schedulers.elastic())
     .timeout(Duration.ofMillis(50))
     .onErrorResume(throwable -> Mono.just("Timeout"))
     .doOnSuccess(result -> logger.info("Result: " + result))
     .block();
}

额外参考：

• https://tacogrammer.com/onerrordropped-explained/

• https://medium.com/@kalpads/configuring-timeouts-in-spring-reactive-webclient-4bc5faf56411