Flux.onErrorContinue() 空对象

Question

我对 onErrorContinue() 有一个问题，即传递给 Biconsumer 的对象为空。

我正在使用 Spring boot 2.1.13.RELEASE 和响应式 mongo，反应堆核心版本为 3.2.15.RELEASE。

当调用数据库以检索具有 id 的记录并使用运算符 switchIfEmtpy() 我使用 Mono.error() 抛出异常并且下游我尝试使用 onErrorContinue( )。

下面的代码解释了这个问题：

public static void main(String[] args) {
    Flux.range(1, 10)
            .flatMap(integer -> mapInteger(integer))
            .doOnNext(System.out::println)
            .onErrorContinue((throwable, o) -> System.out.println("error with " + o)) // o is null
            .subscribe();
}

public static Mono<Integer> mapInteger(Integer num) { // This is here to simulate the db call
    return Mono.just(num)
            .flatMap(t -> {
                if (t == 5)
                    return Mono.empty();
                else
                    return Mono.just(num * 2);
            })
            .switchIfEmpty(Mono.error(new RuntimeException("Error happened while mapping integer!")));
}

这将打印以下值：

2
4
6
8
error with null
12
14
16
18
20

PS。当流中发生另一个错误时，我没有问题。

更新： mapInteger() 用于模拟以下对响应式mongo 存储库的调用：

public Mono<MetaData> getFromDbByKey(String key) {
        return repository
            .findByKeyAndDeletedIsFalse(key)
            .switchIfEmpty(Mono.error(() -> new RuntimeException()));
    }

并且对 getFromDbByKey() 的调用返回这个元数据，我需要将它映射到主流程中的通量。

通过 onErrorContinue，我们捕获 Throwable，并根据其类型对每种类型的错误进行不同的处理。

Answer 1

如果 t==5 ，您不能指望在返回 Mono.empty() 时收到“项目” t 吗？

此代码将打印您需要的内容。

    Flux.range(1, 10)
        .flatMap(integer -> mapInteger(integer))
        .doOnNext(System.out::println)
        .onErrorContinue((throwable, o) -> {
          System.out.println(throwable.getMessage());
        })
        .subscribe();
  }
  public static Mono<Integer> mapInteger(Integer num) { // This is here to simulate the db call
    return Mono.just(num)
        .flatMap(t -> {
          if (t == 5)
            return Mono.empty();
          else
            return Mono.just(num * 2);
        })
        .switchIfEmpty(Mono.error(new RuntimeException("error with " + num)));
  }

打印：

2
4
6
8
error with 5
12
14
16
18
20

在你实际的 mongoDB 调用中，你可以有类似的东西：

public Mono<MetaData> getFromDbByKey(String key) {
        return repository
            .findByKeyAndDeletedIsFalse(key)
            .switchIfEmpty(Mono.error(() -> new RuntimeException("Couldnt find metadata which is not deleted for the key: " + key)));
    }

Answer 2

这是一位同事向我建议的解决方案/解决方法：

public static void main(String[] args) {

    Flux.range(1, 10)
        .flatMap(integer -> mapInteger(integer)
                           .switchIfEmpty(Mono.error(
                                new CustomException("Error happened while mapping integer!", 
                            integer))))
        .doOnNext(System.out::println)
        .onErrorContinue((throwable, o) -> {
            if (CustomException.class.isInstance(throwable))
                System.out.println("Object to blame " + ((CustomException)throwable).getBlamedObject());
            System.out.println("error with " + o);
        })
    .subscribe();
  }

  public static Mono<Integer> mapInteger(Integer num) {
      return Mono.just(num)
          .flatMap(t -> {
              if (t == 5)
                  return Mono.empty();
              else
                  return Mono.just(num * 2);
          });
  }

  public static class CustomException extends RuntimeException {
      private Integer blamedObject;

      public CustomException(String message, Integer blamedObject) {
          super(message);
          this.blamedObject = blamedObject;
      }

      public Integer getBlamedObject() {
          return blamedObject;
      }
  }

这将返回：

2
4
6
8
Object to blame 5
error with null
12
14
16
18
20

解决方案/解决方法是创建一个自定义异常来接受有错误的对象并将 switchIfEmpty() 移动到主要流程，以便我可以获取导致问题的对象以及以后在 onErrorContinue() 中，我可以检查异常类型并获取导致问题的对象。

这对我有用，因为示例中的 mapInteger() 只是对响应式 mongo 存储库的模拟，如果在原始帖子中没有找到更新的结果，它将返回空单声道。