SpringBoot 2.7.14批处理如何实现全流程事务控制？

针对你的场景，可以实现全流程事务控制，让所有数据库更新仅在整个批处理执行成功时才提交。但需要注意：CSV文件写入是不可回滚的磁盘操作，必须通过临时文件+Job监听器的方式保证文件输出的原子性，再配合Spring Batch的事务配置实现数据库操作的全局事务。

核心思路 #

1. CSV写入原子化：先将内容写入临时文件，待整个Job执行成功后再重命名为目标CSV；若Job失败则删除临时文件，避免生成不完整的CSV。
2. 数据库操作全局事务：将整个Step的所有Chunk执行逻辑纳入同一个事务，所有数据库更新操作在事务中累积，仅当全量处理成功时才提交事务，任意环节失败则回滚所有更新。

具体实现步骤 #

1. 实现临时文件版CSV Writer #

替换默认的FlatFileItemWriter，先写入临时文件，后续通过监听器处理最终文件生成：

@Component
public class TempFileCsvItemWriter<T> extends FlatFileItemWriter<T> {

    private String targetFilePath;
    private String tempFilePath;

    @Override
    public void afterPropertiesSet() throws Exception {
        // 生成临时文件路径（目标路径后追加.tmp后缀）
        this.tempFilePath = targetFilePath + ".tmp";
        setResource(new FileSystemResource(tempFilePath));
        super.afterPropertiesSet();
    }

    public void setTargetFilePath(String targetFilePath) {
        this.targetFilePath = targetFilePath;
    }

    public String getTempFilePath() {
        return tempFilePath;
    }
}

2. 配置Job监听器处理临时文件 #

在Job执行完成后，根据执行状态决定临时文件的保留或删除：

@Component
public class CsvJobCompletionListener implements JobExecutionListener {

    @Autowired
    private TempFileCsvItemWriter<?> csvItemWriter;

    @Override
    public void afterJob(JobExecution jobExecution) {
        File tempFile = new File(csvItemWriter.getTempFilePath());
        if (!tempFile.exists()) return;

        if (jobExecution.getStatus() == BatchStatus.COMPLETED) {
            // 执行成功：将临时文件重命名为目标CSV
            File targetFile = new File(csvItemWriter.getTargetFilePath());
            boolean renameSuccess = tempFile.renameTo(targetFile);
            if (!renameSuccess) {
                throw new RuntimeException("Failed to rename temp CSV file to target path");
            }
        } else {
            // 执行失败：删除临时文件
            tempFile.delete();
        }
    }
}

3. 配置全局事务的Step与Job #

修改Batch配置，让整个Step在单一事务中执行，并组合CompositeItemWriter：

@Configuration
@EnableBatchProcessing
public class BatchConfig {

    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;
    private final EntityManagerFactory entityManagerFactory;
    private final TempFileCsvItemWriter<YourEntity> csvItemWriter;
    private final CsvJobCompletionListener jobCompletionListener;

    public BatchConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory,
                       EntityManagerFactory entityManagerFactory, TempFileCsvItemWriter<YourEntity> csvItemWriter,
                       CsvJobCompletionListener jobCompletionListener) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.entityManagerFactory = entityManagerFactory;
        this.csvItemWriter = csvItemWriter;
        this.jobCompletionListener = jobCompletionListener;
    }

    // 配置JPA Writer（用于数据库更新）
    @Bean
    public JpaItemWriter<YourEntity> jpaItemWriter() {
        JpaItemWriter<YourEntity> writer = new JpaItemWriter<>();
        writer.setEntityManagerFactory(entityManagerFactory);
        writer.setFlushInterval(1); // 确保每次更新及时触发生命周期
        // 针对更新操作，若实体为托管状态可省略，若为 detached 状态需设置MERGE模式
        // writer.setEntityClass(YourEntity.class);
        return writer;
    }

    // 配置CompositeItemWriter，组合CSV写入与数据库更新
    @Bean
    public CompositeItemWriter<YourEntity> compositeItemWriter() {
        CompositeItemWriter<YourEntity> compositeWriter = new CompositeItemWriter<>();
        compositeWriter.setDelegates(List.of(csvItemWriter, jpaItemWriter));
        return compositeWriter;
    }

    // 配置Reader（查询数据库视图）
    @Bean
    public JpaPagingItemReader<YourEntity> itemReader() {
        JpaPagingItemReader<YourEntity> reader = new JpaPagingItemReader<>();
        reader.setEntityManagerFactory(entityManagerFactory);
        reader.setQueryString("SELECT e FROM YourEntity e"); // 替换为你的视图查询语句
        reader.setPageSize(1); // 与chunk size保持一致
        return reader;
    }

    // 配置Step，启用全局事务
    @Bean
    public Step dataProcessingStep() {
        return stepBuilderFactory.get("dataProcessingStep")
                .<YourEntity, YourEntity>chunk(1)
                .reader(itemReader())
                .writer(compositeItemWriter())
                .transactionManager(transactionManager())
                .transactionAttribute(globalTransactionAttribute())
                .build();
    }

    // 配置JPA事务管理器
    @Bean
    public PlatformTransactionManager transactionManager() {
        JpaTransactionManager transactionManager = new JpaTransactionManager();
        transactionManager.setEntityManagerFactory(entityManagerFactory);
        return transactionManager;
    }

    // 配置全局事务属性，确保整个Step在同一事务中执行
    @Bean
    public TransactionAttribute globalTransactionAttribute() {
        DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
        attribute.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRED);
        attribute.setTimeout(300); // 根据数据量调整事务超时时间
        return attribute;
    }

    // 配置Job，绑定监听器
    @Bean
    public Job dataProcessingJob() {
        return jobBuilderFactory.get("dataProcessingJob")
                .incrementer(new RunIdIncrementer())
                .listener(jobCompletionListener)
                .flow(dataProcessingStep())
                .end()
                .build();
    }
}

关键注意事项 #

• 数据量限制：全局事务会持有数据库连接直到整个Job完成，若处理数据量极大，可能导致事务超时或连接池耗尽，需根据实际场景评估是否适用。
• 临时文件存储：确保临时文件与目标文件在同一文件系统下，否则renameTo可能失败；跨文件系统时需改用文件复制+删除临时文件的逻辑。
• JPA实体状态：Reader查询的实体需为JPA托管状态（JpaPagingItemReader默认满足），否则JPA Writer需显式设置MERGE模式来处理更新。

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