Spring Cloud Task+Spring Batch声明式事务管理失效问题排查

环境 #

Spring Boot 2.5.4

问题描述 #

服务层和DAO层均已添加@Transactional注解，但同一服务层方法内的连续DAO调用会各自提交到数据库，无法在同一事务中执行。预期仅当服务层方法无错误完成时才提交事务，但目前DAO操作执行后就直接提交了。

相关配置与代码 #

Spring Cloud Task配置 #

@EnableTask
@SpringBootApplication
@EnableBatchProcessing
@Import(CoreConfiguration.class)
public class SpringBatchApplication {

    public static void main(String[] args) {
        SpringApplication.run(SpringBatchApplication.class, args);
    }

}

应用数据源配置 #

@Configuration
public class CoreConfiguration {

    @Primary
    @Bean({ "appDatasourceProperties" })
    @ConfigurationProperties(prefix = "app.datasource")
    public DataSourceProperties appDatasourceProperties() {
        return new DataSourceProperties();
    }

    @Primary
    @Bean({ "appDataSource" })
    public DataSource appDataSource() {
        return this.appDatasourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();
    }

    @Primary
    @Bean({ "transactionManager" })
    public TransactionManager appTransactionManager() {
        return new DataSourceTransactionManager(appDataSource());
    }

}

SCDF服务器数据库配置 #

@Configuration
@Import(TaskConfiguration.class)
public class DatasourceConfiguration extends DefaultBatchConfigurer {

  @Autowired
  @Qualifier("springCloudDataSource")
  private DataSource springCloudDataSource;
  
  @Override
  @Autowired
  public void setDataSource(@Qualifier("springCloudDataSource") DataSource springCloudDataSource) {
    super.setDataSource(springCloudDataSource);
  }

  @Bean
  public BatchDataSourceInitializer batchDataSourceInitializer(
      @Qualifier("springCloudDataSource") DataSource springCloudDataSource,
      ResourceLoader resourceLoader,
      BatchProperties properties) {
    return new BatchDataSourceInitializer(springCloudDataSource, resourceLoader, properties);
  }

  @Override
  protected JobRepository createJobRepository() throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(springCloudDataSource);
    factory.setTransactionManager(super.getTransactionManager());
    factory.setIsolationLevelForCreate("ISOLATION_READ_COMMITTED");
    return factory.getObject();
  }
}

@Configuration
public class TaskConfiguration {

  @Bean({"springCloudDatasourceProperties"})
  @ConfigurationProperties(prefix = "spring.datasource")
  public DataSourceProperties springCloudDatasourceProperties() {
    return new DataSourceProperties();
  }

  @Bean(name = {"springCloudDataSource"})
  public DataSource springCloudDatasource() {
    return this.springCloudDatasourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();
  }

  @Bean
  public TaskConfigurer taskConfigurer(@Qualifier("springCloudDataSource") DataSource springCloudDatasource) {
    return new DefaultTaskConfigurer(springCloudDatasource);
  }
}

注：这是企业模板，已注意到springCloudDataSource在两处配置中重复定义，暂时认为不影响；为解决问题刚在CoreConfiguration中添加了transactionManager Bean。

服务层代码 #

@Slf4j
@Component
public class FactureRecordingService implements FactureRecording {
    
    @Autowired
    private FactureUploadToAlfresco uploadService;
    
    @Autowired
    private CommonDao common;
    
    @Override
    @Transactional(transactionManager = "transactionManager", propagation = REQUIRES_NEW, rollbackFor = FactureRecordingFailure.class)
    public RecordingResultDTO recordFacture(File pdf, MetaParsingResultDTO meta) throws FactureRecordingFailure {

        String numeroFacture = meta.getOne().get(MetadataEnum.NUMERO_FACTURE);
        RecordingResultDTO result = new RecordingResultDTO();

        try {
            Long factureId = null;
            FacturePdfDTO facturePdf = null;
            
            // 保存PDF发票到数据库
            facturePdf = common.createFacturePDF(meta.getOne());
            factureId = facturePdf.getId();
            result.setFactureDbId(factureId);
            result.setFacture(facturePdf);
    
            // 处理包含多份发票的情况
            if (meta.isHasMany()) {
                for (Map<MetadataEnum, String> facture : meta.getMany()) {
                    common.createDetailFacture(factureId, facture);
                }
            }
    
            // 标记发票已上传，避免重复处理
            common.createSpecificRegistreKey(factureId, meta.getOne().get(MetadataEnum.NUMERO_FACTURE));
    
            log.info("发票 {} 已保存到数据库，ID为 {}",numeroFacture,factureId);
            
            // 上传到GED系统，失败则触发回滚
            Long gedId = uploadService.uploadFacture(pdf,meta.getOne());
            log.info("发票已上传到GED，ID为 {}",gedId);
            
            result.setFactureGedId(gedId);
            
            return result;

        } catch(ParseException pe) {
            throw new FactureRecordingFailure("无法保存发票 "+numeroFacture+"：日期格式错误",pe);
            
        } catch(UploadFailure e) {
            throw new FactureRecordingFailure("发票 "+numeroFacture+" 上传GED失败",e);
            
        } catch(Exception ex) {
            throw new FactureRecordingFailure("保存发票 "+numeroFacture+" 时发生未知错误",ex);
        }
    }

DAO层代码 #

@Component
@Transactional(transactionManager = "transactionManager", propagation = Propagation.REQUIRED)
public class CommonDaoImpl implements CommonDao {
    
    @Autowired
    @Qualifier("appDataSource")
    private DataSource appDataSource;
    
    @Autowired
    private NamedParameterJdbcTemplate template;
    
    @Autowired
    private JdbcTemplate simpleTemplate;
    
    @Autowired
    private FournisseurRules rules;
    
    public FacturePdfDTO createFacturePDF(Map<MetadataEnum,String> data) throws ParseException {
        
        String sql = """
insert into FacturePdf
(typeFacture,numeroFacture,codeFournisseur,dateFacture,typeEnergie,pcl_id,siret,raisonSociale,periodeDebut,periodeFin,montantHT,volume,dateRecuperation,idGed) values 
(:typeFacture,:numeroFacture,:codeFournisseur,:dateFacture,:typeEnergie,:pcl_id,:siret,:raisonSociale,:periodeDebut,:periodeFin,:montantHT,:volume,:dateRecuperation,:idGed) 
                """;

        char typeEnergie = rules.getTypeEnergie().getDbCode();
        String codeFournisseur = rules.getDbCodeFournisseur();
        
        SqlParameterSource namedParameters = new MapSqlParameterSource()
                .addValue("typeFacture",rules.getTypeFacture().getDbCode())
                .addValue("numeroFacture",data.get(MetadataEnum.NUMERO_FACTURE))
                .addValue("codeFournisseur",codeFournisseur)
                .addValue("dateFacture",rules.convertStringToDate(data.get(MetadataEnum.DATE_FACTURE)))
                .addValue("typeEnergie",typeEnergie)
                .addValue("pcl_id", data.get(MetadataEnum.PCL))
                .addValue("siret", data.get(MetadataEnum.SIRET))
                .addValue("raisonSociale",data.get(MetadataEnum.RAISON_SOCIALE))
                .addValue("periodeDebut",rules.convertStringToDate(data.get(MetadataEnum.PERIODE_DEBUT_CONSOMMATION)))
                .addValue("periodeFin",rules.convertStringToDate(data.get(MetadataEnum.PERIODE_FIN_CONSOMMATION)))
                .addValue("montantHT",rules.convertMontantToNumber(data.get(MetadataEnum.MONTANT_HT)))
                .addValue("volume",rules.convertVolumeToNumber(data.get(MetadataEnum.VOLUME)))
                .addValue("dateRecuperation",new Date(System.currentTimeMillis()))
                .addValue("idGed",null);
        
        GeneratedKeyHolder keyHolder = new GeneratedKeyHolder();
        template.update(sql,namedParameters,keyHolder,new String[] { "ID" });

        Long id = keyHolder.getKey().longValue();       
        
        FacturePdfDTO result = new FacturePdfDTO();
        result.setId(id);
        result.setCodeFournisseur(codeFournisseur);
        result.setTypeEnergie(typeEnergie);
        
        return result;
    }
}

实际问题 #

调用createFacturePDF后，DAO插入的数据已被提交，Spring在某处执行了提交操作；预期仅当recordFacture方法无错误完成时才提交事务，不清楚问题根源。

解决方案 #

1. 关闭数据源自动提交 #

HikariDataSource默认开启autoCommit=true，若事务管理器未正确接管连接，JDBC操作会自动提交。在CoreConfiguration的数据源配置中显式关闭自动提交：

@Primary
@Bean({ "appDataSource" })
public DataSource appDataSource() {
    HikariDataSource dataSource = (HikariDataSource) this.appDatasourceProperties().initializeDataSourceBuilder().type(HikariDataSource.class).build();
    dataSource.setAutoCommit(false);
    return dataSource;
}

2. 移除DAO层的@Transactional注解 #

服务层已标注事务注解，DAO层的@Transactional属于冗余配置，可能导致事务代理冲突。移除DAO类上的@Transactional，让服务层事务统一管理所有DAO操作。

3. 开启事务管理注解支持 #

当前配置中未显式开启事务管理，需在CoreConfiguration或启动类上添加@EnableTransactionManagement注解，激活Spring的事务代理功能。

4. 显式绑定JDBC模板与数据源 #

确保NamedParameterJdbcTemplate和JdbcTemplate使用业务数据源，避免Spring自动注入错误的数据源：

@Autowired
public CommonDaoImpl(@Qualifier("appDataSource") DataSource appDataSource) {
    this.template = new NamedParameterJdbcTemplate(appDataSource);
    this.simpleTemplate = new JdbcTemplate(appDataSource);
}

5. 排查Spring Batch事务干扰 #

Spring Batch会创建自身的事务管理器，需确保业务操作使用自定义的transactionManager。当前服务层已指定transactionManager = "transactionManager"，可进一步验证Batch配置未修改业务数据源的事务属性。

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