单读取器多并发处理器的Spring Batch实现方案咨询

你的思路完全合理，批量读取+并发处理是Spring Batch应对百万级数据、缩短处理时间的标准优化方向，下面给出具体可落地的实现方案：

一、核心实现：批量读取+并发处理 #

要实现「一次读100条、5个处理器并发处理」，核心是使用Spring Batch的AsyncItemProcessor+AsyncItemWriter组合，配合线程池控制并发数：

1. 配置异步处理器与线程池 #

// 配置线程池，控制并发数为5
@Bean
public ThreadPoolTaskExecutor batchTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(5);
    executor.setMaxPoolSize(5);
    executor.setThreadNamePrefix("batch-proc-");
    executor.initialize();
    return executor;
}

// 异步处理器，包装你的业务处理器
@Bean
public AsyncItemProcessor<Record, ProcessResult> asyncItemProcessor(ItemProcessor<Record, ProcessResult> businessProcessor) {
    AsyncItemProcessor<Record, ProcessResult> asyncProcessor = new AsyncItemProcessor<>();
    asyncProcessor.setDelegate(businessProcessor);
    asyncProcessor.setTaskExecutor(batchTaskExecutor());
    return asyncProcessor;
}

// 异步写入器，包装你的结果写入逻辑
@Bean
public AsyncItemWriter<ProcessResult> asyncItemWriter(ItemWriter<ProcessResult> resultWriter) {
    AsyncItemWriter<ProcessResult> asyncWriter = new AsyncItemWriter<>();
    asyncWriter.setDelegate(resultWriter);
    return asyncWriter;
}

2. 配置Step实现批量读取 #

使用分页读取器（如JdbcPagingItemReader或JpaPagingItemReader）设置pageSize=100，确保每次读取100条数据：

@Bean
public Step dataProcessingStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                               ItemReader<Record> batchReader,
                               AsyncItemProcessor<Record, ProcessResult> asyncProcessor,
                               AsyncItemWriter<ProcessResult> asyncWriter) {
    return new StepBuilder("dataProcessingStep", jobRepository)
            .<Record, ProcessResult>chunk(100, transactionManager)
            .reader(batchReader)
            .processor(asyncProcessor)
            .writer(asyncWriter)
            .build();
}

二、确保记录不重复处理 #

两种可靠方案，按需选择：

方案1：数据库行锁（推荐） #

在读取SQL中使用SELECT ... FOR UPDATE SKIP LOCKED（MySQL 8+、PostgreSQL等主流数据库支持），读取时锁定记录，其他线程会自动跳过已锁定的数据，从根源避免重复：

SELECT id, field1, field2 FROM target_table WHERE status = 0 FOR UPDATE SKIP LOCKED

方案2：状态标记法 #

给目标表增加status字段（枚举值：0=待处理、1=处理中、2=处理成功、3=处理失败）：

• 读取时只查询status=0的记录
• 读取后立即将记录更新为status=1（需在同一个事务内完成读取+更新，防止并发冲突）
• 处理成功后更新为status=2，失败则更新为status=3并记录失败原因

三、失败记录收集与文件输出 #

通过Spring Batch的监听器机制，捕获失败记录并在作业结束后生成txt文件：

1. 线程安全的失败记录容器 #

// 用线程安全集合存储失败记录，避免并发写入问题
@Component
public class FailureRecordHolder {
    private final ConcurrentLinkedQueue<FailureDetail> failureDetails = new ConcurrentLinkedQueue<>();

    public void addFailure(Long recordId, String reason) {
        failureDetails.add(new FailureDetail(recordId, reason));
    }

    public List<FailureDetail> getFailures() {
        return new ArrayList<>(failureDetails);
    }
}

// 失败记录实体
@Data
@AllArgsConstructor
public class FailureDetail {
    private Long recordId;
    private String failureReason;
}

2. 注册监听器捕获失败 #

@Bean
public ItemProcessListener<Record, ProcessResult> processErrorListener(FailureRecordHolder failureHolder) {
    return new ItemProcessListener<>() {
        @Override
        public void onProcessError(Record item, Exception e) {
            // 记录失败ID和异常原因
            failureHolder.addFailure(item.getId(), e.getMessage());
        }
    };
}

// 作业结束后写入失败文件
@Bean
public JobExecutionListener failureFileWriter(FailureRecordHolder failureHolder) {
    return new JobExecutionListener() {
        @Override
        public void afterJob(JobExecution jobExecution) {
            List<FailureDetail> failures = failureHolder.getFailures();
            if (failures.isEmpty()) return;

            try (BufferedWriter writer = Files.newBufferedWriter(Paths.get("failure_records.txt"))) {
                for (FailureDetail detail : failures) {
                    writer.write(String.format("记录ID: %d, 失败原因: %s%n", detail.getRecordId(), detail.getFailureReason()));
                }
            } catch (IOException e) {
                // 可根据业务需求做日志告警或重试
                e.printStackTrace();
            }
        }
    };
}

// 把监听器注册到Step和Job
@Bean
public Job dataProcessJob(JobRepository jobRepository, Step dataProcessingStep, JobExecutionListener failureFileWriter) {
    return new JobBuilder("dataProcessJob", jobRepository)
            .start(dataProcessingStep)
            .listener(failureFileWriter)
            .build();
}

关键注意事项 #

• 事务边界：如果用状态标记法，建议将单条记录的状态更新放在业务处理器内，确保每条记录的处理状态独立提交，避免因Chunk内部分失败导致全部回滚
• 异常兜底：异步处理器的异常会被封装为ExecutionException，需在监听器中正确解析原始异常信息
• 资源控制：线程池的核心/最大线程数不要设置过高，避免数据库连接池耗尽

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