如何回退项目反应堆通量的错误元素 onErrorReturn？答案

【问题标题】：How can I fall back the wrong element onErrorReturn of a project reactor flux?如何回退项目反应堆通量的错误元素 onErrorReturn？
【发布时间】：2019-02-17 13:38:12
【问题描述】：

假设我有一个元素引发错误的通量。

 Flux.just("aeiou","aeio").map(x -> {
           if(!x.endsWith("u")){throws Exception;}
           return x;})
      .onErrorReturn(/*This only accepts a new element*/)
      .map(x -> x+";")
      .subscribe();

 Flux.just("aeiou","aeio").map(x -> {
           if(!x.endsWith("u")){throws Exception;}
           return x;})
      .onErrorResume(/*This doesn't do the next steps*/)
      .map(x -> x+";")
      .subscribe();

我想要这样的东西

 Flux.just("aeiou","aeio").map(x -> {
           if(!x.endsWith("u")){throws Exception;}
           return x;})
      .onErrorReturn(x -> x + "u")
      .map(x -> x+";")
      .subscribe();

文档似乎有误 http://projectreactor.io/docs/core/release/reference/#_fallback_method

Flux.just("key1", "key2")
   .flatMap(k -> callExternalService(k)) 
   .onErrorResume(e -> getFromCache(k)); //k not resolved here

【问题讨论】：

文档不仅正确，而且是解决您问题的可行解决方案。
@abetteroliver 嗯，我想用 k 的值回退，但是没有解决。我什至粘贴了代码块，解析器不喜欢。
e = Throwable，不是最后一个块中的字符串，k 应该是字符串
文档中的示例在callExternalService创建的流上调用onErrorResume。你的没有。您可以使用 flatMap 而不是 map 来使其工作。

标签： project-reactor reactive

【解决方案1】：

文档是正确的。

 Flux.just("key1", "key2")
   .flatMap(k -> callExternalService(k)) 
   .onErrorResume(e -> getFromCache(k)); //k not resolved here

k 没有得到解决，因为 k 在 flatMap 中被声明为局部变量。它对 onErrorResume() 不可见。如果您在管道上方声明 k，即在 Flux.just 之前，它将被解决。

其次，关于你的另一个问题

Flux.just("aeiou","aeio").map(x -> {
       if(!x.endsWith("u")){throws Exception;}
       return x;})
  .onErrorReturn(x -> x + "u")
  .map(x -> x+";")
  .subscribe();

您必须参考 API 文档。 onErrorReturn() 可以让您返回一个备用值。但是那个备用值是硬编码的，即你不会在那里得到 x 的值。您可以简单地将值硬编码为 u 但这可能无法解决您的目的。

我建议查看onErrorResume(Function<? super Throwable,? extends Publisher<? extends T>> fallback)，因为这允许您编写一个函数，您可以在其中放置一些逻辑来进行额外的处理。

这使您可以取回发布者，因此您可以执行以下操作。

.onErrorResume(e -> { //get the value of x
                       String x = <>;// you will have to write logic to get the value of x here, as it will not be available directly                          
                       return Flux.just(x+"u");
                       })

【讨论】：

谢谢，这是我的问题。我似乎没有找到获得 x 的方法，因为它会随着通量的每一步而变化。
嗯，一种方法是简单地将其传递到 Exception 对象中，然后您可以在 onErrorResume 中检索它。基本上创建一个自定义异常对象，然后在抛出异常之前在该自定义异常对象中设置 x 的值。然后，当您在 onErrorResume 中处理它时，您可以从 Exception 对象中获取 X。
有一些新的东西：onErrorContinue(BiConsumer consumer)。我可以直接捕获有错误和异常的对象。然后通量继续下一个元素，唯一的问题是它不会重新抛出一个新对象。
projectreactor.io/docs/core/release/api/reactor/core/publisher/…