Reactor中的onErrorContinue?和?onErrorResume

更新時(shí)間：2022年09月22日 15:01:13 作者：十二又十三

這篇文章主要介紹了Reactor中的onErrorContinue?和?onErrorResume，文章圍繞主題展開詳細(xì)的內(nèi)容介紹，具有一定的參考價(jià)值，需要的朋友可以參考一下

前言

這似乎是 Reactor 的熱門搜索之一，至少當(dāng)我在谷歌中輸入 onErrorContinue 時(shí)，onErrorResume 會(huì)在它旁邊彈出。讓我把我的測試代碼和我的一些解釋粘貼在下面。

1 基礎(chǔ)功能

這是一個(gè)簡單的函數(shù)，將 5 個(gè)連續(xù)的數(shù)字分別乘以 2，然后相加，當(dāng) i==2 時(shí)拋出一個(gè)異常:

public static void main(String... args) {
      Flux.range(1,5)
              .doOnNext(i -> System.out.println("input=" + i))
              .map(i -> i == 2 ? i / 0 : i)
              .map(i -> i * 2)
              .reduce((i,j) -> i+j)
              .doOnNext(i -> System.out.println("sum=" + i))
              .block();
  }

顯然，輸出如下：

input=1
input=2
Exception in thread "main" java.lang.ArithmeticException: / by zero

2 只有 onErrorResume ()

public static void main(String... args) {
      Flux.range(1,5)
              .doOnNext(i -> System.out.println("input=" + i))
              .map(i -> i == 2 ? i / 0 : i)
              .map(i -> i * 2)
              .onErrorResume(err -> {
                  log.info("onErrorResume");
                  return Flux.empty();
              })
              .reduce((i,j) -> i+j)
              .doOnNext(i -> System.out.println("sum=" + i))
              .block();
  }

輸出如下：

input=1
input=2
17:40:47.828 [main] INFO com.example.demo.config.TestRunner - onErrorResume
sum=2

正如官方文檔描述（onErrorResume）的那樣，onErrorResume 用返回內(nèi)容替換 Flux，因此在 2 之后的數(shù)據(jù)不會(huì)被處理。唯一值得一提的是，onErrorResume() 不必馬上在錯(cuò)誤之后捕獲異常。在 onErrorContinue() 的官網(wǎng)文檔中（onErrorContinue），只有 onErrorContinue() 強(qiáng)調(diào)了了 upstream 關(guān)鍵詞，但顯然，它還有其他含義。

3 只有 onErrorContinue()

public static void main(String... args) {
  Flux.range(1,5)
          .doOnNext(i -> System.out.println("input=" + i))
          .map(i -> i == 2 ? i / 0 : i)
          .map(i -> i * 2)
          .onErrorContinue((err, i) -> {log.info("onErrorContinue={}", i);})
          .reduce((i,j) -> i+j)
          .doOnNext(i -> System.out.println("sum=" + i))
          .block();
}

輸出：

input=1
input=2
17:43:10.656 [main] INFO com.example.demo.config.TestRunner - onErrorContinue=2
input=3
input=4
input=5
sum=26

顯然，onErrorContinue 會(huì)丟棄錯(cuò)誤數(shù)據(jù) 2, 然后繼續(xù)數(shù)字 3 直到 5。

4 onErrorResume() 然后 onErrorContinue()

public static void main(String... args) {
    Flux.range(1,5)
            .doOnNext(i -> System.out.println("input=" + i))
            .map(i -> i == 2 ? i / 0 : i)
            .map(i -> i * 2)
            .onErrorResume(err -> {
                log.info("onErrorResume");
                return Flux.empty();
            })
            .onErrorContinue((err, i) -> {log.info("onErrorContinue={}", i);})
            .reduce((i,j) -> i+j)
            .doOnNext(i -> System.out.println("sum=" + i))
            .block();
}

輸出和上面一樣：

input=1
input=2
17:47:05.789 [main] INFO com.example.demo.config.TestRunner - onErrorContinue=2
input=3
input=4
input=5
sum=26

這樣的結(jié)果，你想到了嗎？onErrorContinue() 會(huì)在 onErrorResume() 得到錯(cuò)誤之前處理這個(gè)錯(cuò)誤。當(dāng)兩個(gè)錯(cuò)誤處理函數(shù)在同一個(gè)函數(shù)中的時(shí)候很明顯，但是當(dāng)你的函數(shù)中只有 onErrorResume()，而一些調(diào)用者實(shí)際上有 onErrorContinue() 時(shí)，你的 onErrorResume() 沒有被調(diào)用的原因可能就不那么明顯了。

5 使用 onErrorResume() 模擬 onErrorContinue()

有些博客建議我們完全不用 onErrorContinue()，且在所有場景中僅用 onErrorResume()。但是上述示例已經(jīng)展示了它們會(huì)產(chǎn)生不同的結(jié)果。那我們怎么實(shí)現(xiàn)呢？

public static void main(String... args) {
    Flux.range(1,5)
            .doOnNext(i -> System.out.println("input=" + i))
            .flatMap(i -> Mono.just(i)
                    .map(j -> j == 2 ? j / 0 : j)
                    .map(j -> j * 2)
                    .onErrorResume(err -> {
                        System.out.println("onErrorResume");
                        return Mono.empty();
                    })
            )
            .reduce((i,j) -> i+j)
            .doOnNext(i -> System.out.println("sum=" + i))
            .block();
}

因此，本質(zhì)上是將可能在 flatMap 或 concatMap 中拋出錯(cuò)誤的操作包裝起來，并在其上使用 onErrorResume()。這樣，它會(huì)產(chǎn)生相同的結(jié)果:

input=1
input=2
onErrorResume
input=3
input=4
input=5
sum=26

6 使用 onErrorResume() 和下游的 onErrorContinue() 模擬 onErrorContinue()

有時(shí)候，onErrorContinue() 放在調(diào)用程序中，您無法控制它。但你仍然需要 onErrorResume()。你該怎么辦?

public static void main(String... args) {
    Flux.range(1,5)
            .doOnNext(i -> System.out.println("input=" + i))
            .flatMap(i -> Mono.just(i)
                    .map(j -> j == 2 ? j / 0 : j)
                    .map(j -> j * 2)
                    .onErrorResume(err -> {
                        System.out.println("onErrorResume");
                        return Mono.empty();
                    })
                    .onErrorStop()
            )
            .onErrorContinue((err, i) -> {log.info("onErrorContinue={}", i);})
            .reduce((i,j) -> i+j)
            .doOnNext(i -> System.out.println("sum=" + i))
            .block();
}

秘訣是在 onErrorResume() 代碼塊的末尾添加 onErrorStop() ——這會(huì)阻塞 onErrorContinue()，這樣它就不會(huì)在 onErrorResume() 之前占用錯(cuò)誤。嘗試刪除 onErrorStop()，你會(huì)看到 onErrorContinue() 在 onErrorResume 之前彈出。

到此這篇關(guān)于Reactor中的onErrorContinue 和 onErrorResume的文章就介紹到這了,更多相關(guān)Reactor onErrorContinue 內(nèi)容請搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: