如何并行调用多验证方法并合并Mono<List<ApiErrorEntity>>结果

核心思路 #

通过重构验证方法为响应式返回，利用Flux实现并行执行所有验证任务，最终合并所有错误列表为单一的Mono<List<ApiErrorEntity>>，兼顾扩展性与响应式编程规范。

关键改进点 #

1. 验证方法响应式改造：将所有验证方法改为返回Mono<List<ApiErrorEntity>>，确保异步操作（如userService.getUserDetails）能正确完成并返回错误。
2. 并行执行与结果合并：用Flux批量处理所有验证任务，并行执行后合并错误列表，新增验证方法只需添加对应Mono即可，无需修改核心逻辑。
3. 上层方法响应式适配：将原同步的validate方法改为响应式，正确处理错误列表并抛出异常。

修改后的完整代码 #

@Service
@RequiredArgsConstructor
@Slf4j
public class ValidationService {

    private final UserService userService;

    public Mono<Boolean> validate(UserRequest userRequest, String empno, ServerHttpRequest request) {
        return validateUserInfo(userRequest, empno, request)
                .flatMap(validationFailures -> {
                    if (!CollectionUtils.isEmpty(validationFailures)) {
                        return Mono.error(new ValidationException(validationFailures));
                    }
                    return Mono.just(true);
                });
    }

    public Mono<List<ApiErrorEntity>> validateUserInfo(UserRequest userRequest, String empno, ServerHttpRequest request) {
        // 收集所有验证任务，新增验证只需在此添加
        List<Mono<List<ApiErrorEntity>>> validationTasks = Arrays.asList(
                validateUserDate(userRequest.flag(), userRequest.enrollDate()),
                validateUser(userRequest, request)
        );

        // 并行执行所有验证，合并错误列表
        return Flux.fromIterable(validationTasks)
                .flatMap(task -> task) // 默认并行度为CPU核心数，可指定参数调整：flatMap(task -> task, 8)
                .collectList()
                .flatMap(errorLists -> {
                    List<ApiErrorEntity> allErrors = new ArrayList<>();
                    errorLists.forEach(allErrors::addAll);
                    return Mono.just(allErrors);
                });
    }

    private Mono<List<ApiErrorEntity>> validateUserDate(Boolean flag, LocalDate enrollDate) {
        List<ApiErrorEntity> errors = new ArrayList<>();
        if (flag == null || !flag) {
            if (enrollDate == null) {
                errors.add(createApiError("regDate", "未提供日期信息"));
            }
            if (enrollDate != null && enrollDate.isBefore(LocalDate.now())) {
                errors.add(createApiError("regDate", "日期不能早于当前时间"));
            }
        }
        return Mono.just(errors);
    }

    private Mono<List<ApiErrorEntity>> validateUser(UserRequest paymentRequest, ServerHttpRequest request) {
        List<ApiErrorEntity> errors = new ArrayList<>();
        return userService.getUserDetails(request)
                .flatMap(userDetails -> {
                    if (!userDetails.getIsActive()) {
                        errors.add(createApiError("userDetails", "用户未激活"));
                    } else if (userDetails.getAge() > 55) {
                        errors.add(createApiError("userDetails", "用户不符合资格"));
                    }
                    return Mono.just(errors);
                })
                .onErrorResume(WebClientResponseException.class, e -> {
                    if (e.getStatusCode().is4xxClientError()) {
                        errors.add(createApiError("userDetails", "无效用户账户"));
                    }
                    return Mono.just(errors);
                });
    }

    private ApiErrorEntity createApiError(String name, String reason) {
        return new ApiErrorEntity(name, reason);
    }
}

说明 #

• 扩展性：新增验证方法时，只需在validationTasks列表中添加对应的Mono<List<ApiErrorEntity>>实现，无需修改其他代码。
• 并行控制：flatMap的第二个参数可指定并行度，比如flatMap(task -> task, 10)表示同时执行10个验证任务。
• 异步正确性：修复了原代码中validateUser方法异步操作未完成就返回错误列表的问题，确保所有错误都被正确捕获。

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