java - Flux 为空时的last()操作

Question

当我需要获取通量的最后一个元素(最后一个方法)时，我试图解决我的问题，但在某些情况下这些通量可能为空并且出现以下错误

Flux#last() didn't observe any onNext signal

这是我的链条

return apiService.getAll(entry)
      .flatMap(response -> {
          if (response.getId() != null){
             //do some logic
             return Mono.just("some Mono");
          }
          else{
             return Mono.empty();
             }
          })
          .last()
          //more flatMap operators

我也已经使用了 switchIfEmpty() 但无法修复。 验证是否可以调用last()或跳过并返回空以终止链操作的正确实现是什么。

谢谢

Answer 1

根据Flux.last() API 文档:

emit NoSuchElementException error if the source was empty. For a passive version use takeLast(int)

这意味着，对于空的上游 Flux:

• last() 将发出错误
• takeLast(1) 将返回一个空通量

现在，takeLast(1) 返回一个 Flux，而不是像 last() 那样返回一个 Mono。然后，您可以将其与 Flux.next() 链接起来，它将返回唯一保留的值(如果有)，或传播空信号。

注意:另一种解决方案是使用 last().onErrorResume(NoSuchElementException.class, err -> Mono.empty())。 这将捕获由 last() 内部发送的错误，然后返回一个空的单声道。 但是，如果除了 last() 之外还有一些可能引发 NoSuchElementException 的代码，您可能会错过问题。为此，我个人对您的情况的选择是使用 takeLast(1).next()。

以下代码示例显示了 last() 与 takeLast(1).next() 的行为:

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class FluxLast {

    static void subscribe(Mono<?> publisher) {
        publisher.subscribe(value -> {},
                            err -> System.out.println("Failed: " + err.getMessage()),
                            () -> System.out.println("Completed empty"));
    }
    
    public static void main(String[] args) {
        subscribe(Flux.empty().last());
        subscribe(Flux.empty().takeLast(1).next());

        // When not empty, takeLast(1).next() will return the last value
        Integer v = Flux.just(1, 2, 3)
                            .takeLast(1)
                            .next()
                            .block();
        System.out.println("Last value: "+v);
    }
}

程序输出:

Failed: Flux#last() didn't observe any onNext signal from Callable flux
Completed empty
3