单声道异步异常处理

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

在你的情况下这是正确的行为。您正在从没有错误的单个字符串“嘿”创建单声道。如果您尝试调试，您​​可以看到该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方法。