Spring Batch+JPA+PostgreSQL事务中止无法写入批次状态如何解决

问题根本原因 #

• PostgreSQL的事务处理规则：同一个事务中任意SQL执行失败后，后续所有SQL都会被拒绝，直到事务结束
• Spring Batch的ItemWriter.write()方法默认运行在Chunk级别的事务中：你调用saveAll()抛出DataIntegrityViolationException时，当前事务已经被标记为回滚状态，此时在catch块中继续调用batchStatusRepository.save()还是同一个事务，自然会触发报错
• 之前尝试加@Transactional不生效的原因：Spring AOP基于代理实现，同类内部方法调用不会触发代理逻辑，所以加在同类方法上的事务注解无效

解决方案 #

方案1：将批次状态写入逻辑放到独立事务中 #

把批次状态的保存逻辑抽为独立类，指定事务传播级别为REQUIRES_NEW，会开启独立新事务，不受原有失败事务的影响：

1. 先定义独立的状态管理服务类

import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

@Service
public class BatchStatusService {

    private final BatchStatusRepository batchStatusRepository;

    public BatchStatusService(BatchStatusRepository batchStatusRepository) {
        this.batchStatusRepository = batchStatusRepository;
    }

    // 开启独立新事务，和原有的Writer事务完全隔离
    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
    public void saveBatchStatus(Long batchId, String status) {
        batchStatusRepository.save(BatchStatusEntity.builder()
                .batchId(batchId)
                .status(status)
                .build());
    }
}

2. 修改CsvWriter注入这个服务，替换原来的直接保存逻辑

@JobScope
@Component
@Slf4j
public class CsvWriter implements ItemWriter<List<CsvEntity>> {

    private final DataRepository dataRepository;
    private final BatchStatusService batchStatusService;

    // 构造器注入
    public CsvWriter(DataRepository dataRepository, BatchStatusService batchStatusService) {
        this.dataRepository = dataRepository;
        this.batchStatusService = batchStatusService;
    }

    @Override
    public void write(List<? extends List<DataEntity>> data) throws Exception {
        Long batchId = data.get(0).get(0).getBatchId();
        try {
            dataRepository.saveAll(data.get(0));
            batchStatusService.saveBatchStatus(batchId, "Ok");
        } catch (DataIntegrityViolationException e) {
            log.error("Duplicate entity", e);
            batchStatusService.saveBatchStatus(batchId, "Error");
            // 如果需要整个步骤标记为失败，这里可以继续抛出异常触发Spring Batch回滚逻辑
            // throw e;
        }
    }
}

方案2：使用Spring Batch原生监听机制写入状态（更符合规范） #

不用在Writer里处理异常，直接利用StepExecutionListener在步骤执行结束后统一写入状态，此时Chunk事务已经完成，不会有事务冲突问题：

import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.stereotype.Component;

@Component
public class BatchStatusListener implements StepExecutionListener {

    private final BatchStatusService batchStatusService;

    public BatchStatusListener(BatchStatusService batchStatusService) {
        this.batchStatusService = batchStatusService;
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        // 从Step上下文中获取你之前存入的batchId
        Long batchId = (Long) stepExecution.getExecutionContext().get("batchId");
        String status = stepExecution.getExitStatus().getExitCode().equals(ExitStatus.COMPLETED.getExitCode()) ? "Ok" : "Error";
        batchStatusService.saveBatchStatus(batchId, status);
        return stepExecution.getExitStatus();
    }
}

然后在你的Step定义中注册这个监听器即可，记得在Step执行前把batchId存入Step的ExecutionContext中。

注意事项 #

• 如果业务数据写入失败后需要整个批次回滚，方案1中catch块处理完状态写入后需要重新抛出异常，触发Spring Batch的原有回滚逻辑
• 不建议在ItemWriter中吞异常不抛出，会导致Spring Batch认为该Chunk执行成功，不会触发重试/跳过等配置逻辑

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