Java Project Reactor ParallelFlux异常忽略优雅方案咨询

针对你遇到的ParallelFlux处理异常导致整个Sink停止工作的问题，Reactor库提供了多个内置操作符来优雅地忽略单个元素的异常，避免终止整个序列，以下是具体方案：

1. 使用onErrorContinue跳过异常元素继续处理 #

onErrorContinue是最直接的方案，它会捕获处理过程中抛出的异常，跳过出错的元素，继续处理后续的消息，同时允许你记录异常信息：

@PostConstruct
private void init(){
    sinks.putIfAbsent(getCode(), Sinks.many().unicast().onBackpressureBuffer());
    sinks.get(getCode()).asFlux()
        .parallel()
        .runOn(scheduler)
        .doOnNext(this::handleMessage)
        .onErrorContinue((throwable, message) -> {
            // 这里可以添加异常日志记录，比如：
            // log.error("处理消息失败，消息内容: {}", message, throwable);
        })
        .subscribe();
}

说明：

• onErrorContinue会拦截doOnNext中抛出的所有异常，不会终止整个ParallelFlux序列，后续消息仍会正常处理。
• 操作符的位置很关键，必须放在doOnNext之后，才能精准捕获消息处理阶段的异常。

2. 使用onErrorResume实现细粒度异常恢复 #

如果需要对异常做更灵活的处理（比如根据异常类型选择不同的恢复逻辑），可以将消息处理逻辑包装为Mono，配合onErrorResume返回空流来忽略异常：

@PostConstruct
private void init(){
    sinks.putIfAbsent(getCode(), Sinks.many().unicast().onBackpressureBuffer());
    sinks.get(getCode()).asFlux()
        .parallel()
        .runOn(scheduler)
        .flatMap(message -> 
            Mono.fromRunnable(() -> handleMessage(message))
                .onErrorResume(throwable -> {
                    // 记录异常或执行自定义恢复逻辑
                    // log.error("处理消息出错", throwable);
                    return Mono.empty();
                })
        )
        .subscribe();
}

说明：

• 这种方式将每个消息的处理独立包装为Mono，单个消息的异常只会终止当前Mono，不会影响其他并行流的处理。
• onErrorResume可以根据异常类型返回不同的恢复流，灵活性更高。

对比手动try/catch的优势 #

• 符合Reactor声明式编程风格，代码更简洁易读，无需在业务逻辑中嵌入大量try/catch块。
• 可以集中处理所有异常，统一日志记录逻辑，避免重复代码。

内容的提问来源于stack exchange，提问作者alext