java - Spring react 器中的错误处理未按预期工作

Question

我无法弄清楚为什么即使 tryDemo 方法抛出 NPE 也没有执行 doOnError 代码。我正在尝试了解响应式(Reactive)编程中的错误处理，我是新手

Mono.zip( Mono.fromCallable( ()->tryDemo()), Mono.fromCallable( ()-   >tryDemo1()),Mono.fromCallable(()-> tryDemo2() ))
        .flatMap( data -> Mono.just( Tuples.of( data.getT1(), data.getT2(),data.getT3() ) ) )
        .doOnError( e -> log.error( "Error {}", e.getStackTrace() ) )
        .subscribe(T->{log.info("Tuple {}",T.getT2()  );});



public Mono<String> tryDemo() {
    log.info( "Data--1" );

    return Mono.error( NullPointerException::new );
    //return Mono.just( "1" );
}

15:56:07.023 [main] INFO com.infosys.rtbm.Test - Data--1
15:56:07.027 [main] INFO com.infosys.rtbm.Test - Tuple MonoJust

Answer 1

别忘了订阅我

在您的示例中，您返回 Mono的错误。这里棘手的部分是 fromCallable期望您返回一个标量值。

如果我们看一下 Mono.fromCallable 的 API ，我们会发现接受的参数是

public static <T> Mono<T> fromCallable(Callable<? extends T> supplier)

这意味着如果我们的 Callable返回 Mono , 我们将得到

Mono<Mono<Object>> monoOfMono = Mono.fromCallable(() -> 
     Mono.error(NullPointerException::new)
);

因此，如果我们需要产生错误，我们必须直接在 lambda 中抛出该异常

Mono<Object> justMono = Mono.fromCallable(() -> {
     throws new NullPointerException()
});

总结，Mono.fromCallable不会尝试检查返回类型是否为流。因此，您的 Mono被认为是正常的标量值并向下游传播。因此，要解决此问题，您可以执行以下操作:

在您的 tryDemo 中抛出异常方法:

Mono.zip( Mono.fromCallable( ()->tryDemo()), Mono.fromCallable( ()-   >tryDemo1()),Mono.fromCallable(()-> tryDemo2() ))
    .flatMap( data -> Mono.just( Tuples.of( data.getT1(), data.getT2(),data.getT3() ) ) )
    .doOnError( e -> log.error( "Error {}", e.getStackTrace() ) )
    .subscribe(T->{log.info("Tuple {}",T.getT2()  );});



public String tryDemo() {
    log.info( "Data--1" );

    throw new NullPointerException();
    //return "1";
}

替换fromCallable到 defer

你可能会变得懒惰，这是你想用 fromCallable 实现的, 使用 Mono.defer在那种情况下期望的运算符 Mono作为 lambda 的返回类型。

如果我们查看该运算符的 API，我们将观察到以下内容

public static <T> Mono<T> defer(Supplier<? extends Mono<? extends T>> supplier);

在这种情况下，我们有 Supplier这正是 Mono将某些东西作为返回类型，所以一旦您再次尝试您的初始代码，您将实现预期的行为:

Mono<Object> justMono = Mono.defer(() -> 
     Mono.error(NullPointerException::new)
);

在这种情况下，一旦供应商返回 Mono , Mono.defere订阅它并收到错误信号:

Mono.zip( 
        Mono.defer(() -> tryDemo()), 
        Mono.defer(() -> tryDemo1()),
        Mono.defer(() -> tryDemo2())
    )
    .flatMap(data -> Mono.just(Tuples.of( 
         data.getT1(), 
         data.getT2(), 
         data.getT3()
    )))
    .doOnError( e -> log.error( "Error {}", e.getStackTrace() ) )
    .subscribe(T -> {
        log.info("Tuple {}",T.getT2()  );
    });