从 Flux<Void> 收集所有错误

【问题标题】：collect all errors from a Flux<Void>从 Flux<Void> 收集所有错误
【发布时间】：2021-11-24 13:47:21
【问题描述】：

我有一个Flux<Message> 和一个Mono<Void> process(Message m) 函数。 process 函数可以返回错误或不返回任何内容。我想并行调用process函数，将所有有进程错误的messageId收集到List<String>中。

@Data
class Message {
    String messageId;
    String content;    
}

Flux.just(new Message("A", "1"), new Message("B", "2"))
    .flatMap(m -> 
        process(m).onErrorReturn(m.messageId)
    )

但问题是我无法在 onErrorXXX 方法中返回 Void 以外的类型。

【问题讨论】：

标签： java reactive-programming spring-webflux project-reactor reactor

【解决方案1】：

你可以做的是有一个包含 messageId 和错误消息的包装器，如下所示：

@Value
private static class ProcessingError {
    private final String id;
    private final String message;
    private final boolean hasError;
}

Flux.just(new Message("A", "1"), new Message("B", "2"))
    .flatMap(m -> process(m).thenReturn(new ProcessingError(m.messageId, null, false))
                            .onErrorResume(th -> Mono.just(new ProcessingError(m.messageId, th.getMessage(), true))));

现在flatMap 之后的通量是Flux<ProcessingError>，您可以只过滤带有hasError = true 的通量，然后以您想要的方式收集。

【讨论】：