基于Flux/Executor Service实现批量客户数据并行更新方案咨询

一、问题背景 #

现有单线程循环处理1000个客户号码，每个号码需先检查数据库是否存在，不存在则调用WebClient查询API并持久化，单线程耗时1000秒，目标通过10线程并行将耗时压缩至100秒。此前尝试@Async结合Executor未生效，现提供Executor线程池与Reactor Flux两种并行实现方案，并对比优劣。

二、Executor线程池方案实现 #

1. 配置指定大小的线程池 #

@Configuration
public class AsyncThreadPoolConfig {
    @Bean(name = "customerProcessExecutor")
    public Executor customerProcessExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(10);
        executor.setMaxPoolSize(10);
        executor.setQueueCapacity(100);
        executor.setThreadNamePrefix("Customer-Process-");
        executor.initialize();
        return executor;
    }
}

2. 抽离异步处理单条号码逻辑 #

注意：异步方法需与调用方法分属不同类，避免Spring AOP代理失效导致单线程执行。

@Service
public class SingleCustomerProcessor {
    private final CustomerClient customerClient;
    private final DbService dbService;

    public SingleCustomerProcessor(CustomerClient customerClient, DbService dbService) {
        this.customerClient = customerClient;
        this.dbService = dbService;
    }

    @Async("customerProcessExecutor")
    public CompletableFuture<Void> processCustomer(String number) {
        if (!isExistBy(number)) {
            String response = customerClient.getCustomerRetrieveApi(number);
            dbService.persistCustomerDetails(response);
        }
        return CompletableFuture.completedFuture(null);
    }

    private boolean isExistBy(String number) {
        // 原有数据库存在性检查逻辑
        return false;
    }
}

3. 批量触发并行任务 #

@Service
public class BatchCustomerService {
    private final SingleCustomerProcessor customerProcessor;

    public BatchCustomerService(SingleCustomerProcessor customerProcessor) {
        this.customerProcessor = customerProcessor;
    }

    public void updateCustomerDetailsInDB(List<String> numberList) {
        List<CompletableFuture<Void>> futures = numberList.stream()
                .map(customerProcessor::processCustomer)
                .collect(Collectors.toList());

        // 等待所有并行任务执行完成
        CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
    }
}

三、Reactor Flux方案实现 #

基于响应式编程改造，移除WebClient的block()调用，实现非阻塞并行。

1. 改造CustomerClient的API调用（返回Mono而非阻塞结果） #

public Mono<String> getCustomerRetrieveApi(String number) {
    // 构造requestObject、urlString等逻辑
    return webClient.post()
            .uri(urlString)
            .contentType(mediaType)
            .bodyValue(requestObject)
            .retrieve()
            .bodyToMono(String.class)
            .onErrorResume(WebClientResponseException.class,
                    ex -> ex.getRawStatusCode() == 404 ? Mono.empty() : Mono.error(ex));
}

2. 并行批量处理实现 #

@Service
public class ReactorBatchCustomerService {
    private final CustomerClient customerClient;
    private final DbService dbService;

    public ReactorBatchCustomerService(CustomerClient customerClient, DbService dbService) {
        this.customerClient = customerClient;
        this.dbService = dbService;
    }

    public void updateCustomerDetailsInDB(List<String> numberList) {
        Flux.fromIterable(numberList)
                // 指定并行度为10
                .parallel(10)
                // 绑定到弹性线程池，避免阻塞事件循环
                .runOn(Schedulers.boundedElastic())
                // 过滤已存在的客户
                .filter(number -> !isExistBy(number))
                // 非阻塞调用API并持久化
                .flatMap(number -> customerClient.getCustomerRetrieveApi(number)
                        .flatMap(response -> Mono.fromRunnable(() -> dbService.persistCustomerDetails(response))))
                // 恢复串行流，等待所有任务完成
                .sequential()
                .blockLast();
    }

    private boolean isExistBy(String number) {
        // 原有数据库存在性检查逻辑
        return false;
    }
}

四、方案对比 #

五、关键注意事项 #

1. Executor方案中，@Async方法不能被同一类内方法调用，否则Spring AOP代理不生效，仍为单线程。
2. Flux方案中，若数据库持久化为阻塞操作（如JDBC），需用Mono.fromRunnable绑定到弹性线程池，避免阻塞响应式事件循环。
3. 两种方案均需调整数据库连接池配置，确保并行操作时不会耗尽连接数。

内容的提问来源于stack exchange，提问作者Shanmuga PriyaV sp