Webflux Reactor：如何判断原Flux中所有请求是否全部成功？

需求：针对一组accountId，依次执行两个请求——先调用接口删除账户数据，仅当第一个请求成功时，再触发对应事件；最终需要确认所有账户的这组请求是否全部成功。

当前实现代码：

Flux.fromIterable(List.of("accountId", "someOtherAccountId"))
            .flatMap(accountId -> someWebclient.deleteAccountData(accountId)
                .doOnSuccess(response -> log.info("Delete account data success"))
                .onErrorResume(e -> {
                    log.info("Delete account data failure");
                    return Mono.empty();
                })
                .flatMap(deleteAccountDataResponse -> {
                    return eventServiceClient.triggerEvent("deleteAccountEvent")
                        .doOnSuccess(response -> log.info("Delete account event success"))
                        .onErrorResume(e -> {
                            log.info("Delete account event failure");
                            return Mono.empty();
                        });
                }))
            .count()
            .subscribe(items -> {
                if (items.intValue() == accountIdsToForget.size()) {
                    log.info("All accountIds deleted and events triggered successfully");
                } else {
                    log.info("Not all accoundIds deleted and events triggered successfully");
                }
            });

当前实现的痛点：用onErrorResume吞掉错误导致丢失失败细节，只能通过count对比判断整体结果，无法定位具体失败的账户。

优化实现方案 #

你的实现有几个可以优化的点，核心是保留错误细节和明确每个账户的处理状态，以下是更符合Reactor惯用风格的实现：

1. 用结果对象封装处理状态 #

先定义一个简单的结果类，用来记录每个账户的处理结果（成功/失败、错误原因）：

import lombok.Data;

@Data
public class AccountProcessingResult {
    private String accountId;
    private boolean success;
    private String errorMessage;

    // 静态工厂方法简化创建
    public static AccountProcessingResult success(String accountId) {
        AccountProcessingResult result = new AccountProcessingResult();
        result.setAccountId(accountId);
        result.setSuccess(true);
        return result;
    }

    public static AccountProcessingResult failure(String accountId, String errorMessage) {
        AccountProcessingResult result = new AccountProcessingResult();
        result.setAccountId(accountId);
        result.setSuccess(false);
        result.setErrorMessage(errorMessage);
        return result;
    }
}

2. 重构Reactor处理逻辑 #

List<String> accountIds = List.of("accountId", "someOtherAccountId");

Flux.fromIterable(accountIds)
    // 可选：控制并发数，避免请求过载，比如限制同时处理5个账户
    .flatMap(accountId -> 
        // 第一步：执行删除账户数据请求
        someWebclient.deleteAccountData(accountId)
            .doOnSuccess(response -> log.info("账户[{}]数据删除成功", accountId))
            // 删除失败时，直接返回失败结果
            .onErrorResume(e -> {
                log.error("账户[{}]数据删除失败：{}", accountId, e.getMessage());
                return Mono.just(AccountProcessingResult.failure(accountId, "删除数据失败：" + e.getMessage()));
            })
            // 删除成功后，触发事件
            .flatMap(deleteResponse -> 
                eventServiceClient.triggerEvent("deleteAccountEvent")
                    .doOnSuccess(response -> log.info("账户[{}]事件触发成功", accountId))
                    // 事件触发失败时，返回失败结果
                    .onErrorResume(e -> {
                        log.error("账户[{}]事件触发失败：{}", accountId, e.getMessage());
                        return Mono.just(AccountProcessingResult.failure(accountId, "触发事件失败：" + e.getMessage()));
                    })
                    // 事件触发成功，返回成功结果
                    .map(response -> AccountProcessingResult.success(accountId))
            )
    , 5)
    // 收集所有账户的处理结果
    .collectList()
    .subscribe(results -> {
        boolean allSuccess = results.stream().allMatch(AccountProcessingResult::isSuccess);
        if (allSuccess) {
            log.info("所有账户均处理成功");
        } else {
            // 提取失败账户详情，便于排查
            List<String> failedDetails = results.stream()
                .filter(result -> !result.isSuccess())
                .map(result -> String.format("账户[%s]：%s", result.getAccountId(), result.getErrorMessage()))
                .toList();
            log.warn("部分账户处理失败，详情：{}", failedDetails);
        }
    });

优化点说明 #

• 不吞错误，保留细节：用onErrorResume返回失败结果而非Mono.empty()，每个账户的失败原因都能被记录下来。
• 明确成功条件：只有删除数据+触发事件两步都成功，才算该账户处理完成，任何一步失败都会标记为整体失败。
• 更灵活的结果判断：通过collectList()收集所有结果，既能判断是否全部成功，也能直接定位具体失败的账户和原因。
• 可控的并发数：给flatMap添加第二个参数限制并发量，避免大量请求同时发起导致服务过载。

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