单声道异步异常处理

Question

我只是想弄清楚异常处理在 reactor 库中是如何工作的。

考虑以下示例：

public class FluxTest {

@Test
public void testIt() throws InterruptedException {
    Scheduler single = Schedulers.single();

    CountDownLatch latch = new CountDownLatch(1);

    Mono.just("hey")
            .subscribeOn(single)
            .doOnError(e -> System.out.println("ERROR1: " + e.getMessage()))
            .doOnTerminate((r, e) -> {
                if (e != null) System.out.println("ERROR3: " + e.getMessage());
                latch.countDown();
            })
            .subscribe(
                    it -> { throw new RuntimeException("Error for " + it); },
                    ex -> { System.out.println("ERROR2: " + ex.getMessage());}
                    );

    latch.await();
}

}

实际上，我看不到任何错误处理代码块的执行。测试终止，没有任何消息。另外，我尝试删除 doOnError, doOnTerminate 个处理器，但没有成功。

Answer 1

在你的情况下这是正确的行为。您正在从没有错误的单个字符串 "hey" 创建单声道。如果您尝试调试，您会看到 doOnTerminate 方法是使用 e = null 参数调用的，根据文档，它在任何情况下都会被调用 success 或 error.

要测试一些错误处理，您可以做下一件事：

public class FluxTest {
    @Test
    public void testIt() throws InterruptedException {
        Scheduler single = Schedulers.single();

        CountDownLatch latch = new CountDownLatch(1);

        Mono.just("hey")
                .map(this::mapWithException)
                .subscribeOn(single)
                .doOnError(e -> System.out.println("ERROR1: " + e.getMessage()))
                .doOnTerminate((r, e) -> {
                    if (e != null) System.out.println("ERROR3: " + e.getMessage());
                    latch.countDown();
                })
                .subscribe(
                        it -> { throw new RuntimeException("Error for " + it); },
                        ex -> { System.out.println("ERROR2: " + ex.getMessage());}
                        );

        latch.await();
    }

    private String mapWithException(String s) {
        throw new RuntimeException();
    }
}

在运行使用上面的代码测试后，您应该在控制台中输入三行代码

ERROR1: null
ERROR3: null
ERROR2: null

所以当单声道失败时第一个回调是 onError，第二个是 onTerminate 因为单声道因错误而终止，第三个 errorConsumer 来自 subscribe 方法。

单声道异步异常处理

Mono async exception handling

java

project-reactor