Spring Integration与Spring Batch事务管理器冲突及文件重复处理问题

核心问题分析 #

当前冲突源于：JdbcMetadataStore需要事务保证putIfAbsent的原子性（避免多实例重复处理），但Spring Batch的JobRepository不允许外部事务注入；若去掉事务，PostgreSQL的唯一约束会因多实例并发插入触发冲突，导致重复处理。

方案一：为JdbcMetadataStore与Spring Batch配置独立事务管理器 #

将两者的数据源、事务管理器完全隔离，避免事务上下文互相干扰。

配置示例 #

// MetadataStore专属数据源与事务管理器
@Bean
public DataSource metadataStoreDataSource() {
    HikariDataSource ds = new HikariDataSource();
    ds.setJdbcUrl("jdbc:postgresql://localhost:5432/metadata_db");
    ds.setUsername("db_user");
    ds.setPassword("db_pass");
    return ds;
}

@Bean
public PlatformTransactionManager metadataTxManager() {
    return new DataSourceTransactionManager(metadataStoreDataSource());
}

@Bean
public JdbcMetadataStore jdbcMetadataStore() {
    JdbcMetadataStore store = new JdbcMetadataStore(metadataStoreDataSource());
    store.setTransactionManager(metadataTxManager());
    return store;
}

// Spring Batch专属数据源与事务管理器
@Bean
public DataSource batchDataSource() {
    HikariDataSource ds = new HikariDataSource();
    ds.setJdbcUrl("jdbc:postgresql://localhost:5432/batch_db");
    ds.setUsername("db_user");
    ds.setPassword("db_pass");
    return ds;
}

@Bean
public PlatformTransactionManager batchTxManager() {
    return new DataSourceTransactionManager(batchDataSource());
}

@Bean
public JobRepository jobRepository() throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(batchDataSource());
    factory.setTransactionManager(batchTxManager());
    factory.setDatabaseType("POSTGRES");
    return factory.getObject();
}

优缺点 #

• 优点：彻底隔离事务上下文，无冲突风险；
• 缺点：需维护两个独立数据源，配置复杂度略高。

方案二：调整事务边界，限制JdbcMetadataStore的事务范围 #

不给整个Integration流添加全局事务，仅为SftpPersistentAcceptOnceFileListFilter配置专属事务管理器，让putIfAbsent操作在独立事务中完成，避免事务传递到后续的JobLaunchingGateway。

配置示例 #

@Bean
public IntegrationFlow sftpInboundFlow() {
    return IntegrationFlows.from(Sftp.inboundAdapter(sftpSessionFactory())
                    .remoteDirectory("/remote/files")
                    .filter(new SftpPersistentAcceptOnceFileListFilter(jdbcMetadataStore(), "sftp_file_region")
                            // 仅给过滤器配置事务管理器
                            .setTransactionManager(metadataTxManager())),
                    e -> e.poller(Pollers.fixedDelay(1000)))
            .transform(fileToJobLaunchRequestTransformer())
            // 给Batch作业单独配置事务，使用PROPAGATION_REQUIRES_NEW开启新事务
            .handle(JobLaunchingGateway(jobLauncher()), 
                    e -> e.advice(batchTxAdvice()))
            .get();
}

@Bean
public TransactionInterceptor batchTxAdvice() {
    TransactionInterceptor advice = new TransactionInterceptor();
    advice.setTransactionManager(batchTxManager());
    DefaultTransactionAttribute txAttr = new DefaultTransactionAttribute();
    txAttr.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    advice.setTransactionAttributes(Collections.singletonMap("*", txAttr));
    return advice;
}

优缺点 #

• 优点：无需额外数据源，仅通过事务传播行为控制范围；
• 缺点：需精准控制事务边界，避免事务意外传递。

方案三：利用PostgreSQL原生语法优化JdbcMetadataStore原子性 #

PostgreSQL支持INSERT ... ON CONFLICT DO NOTHING语法，可替代事务实现putIfAbsent的原子性：多实例并发插入时，只有第一个成功，其余自动忽略，既保证了唯一性，又无需事务，从根源避免与Spring Batch的事务冲突。

配置示例 #

@Bean
public JdbcMetadataStore jdbcMetadataStore(DataSource dataSource) {
    JdbcMetadataStore store = new JdbcMetadataStore(dataSource);
    // 重写putIfAbsent的SQL，使用PostgreSQL的ON CONFLICT语法
    String putSql = """
        INSERT INTO INT_METADATA_STORE (METADATA_KEY, METADATA_VALUE, REGION)
        VALUES (?, ?, ?) ON CONFLICT (METADATA_KEY, REGION) DO NOTHING
        """;
    store.setPutIfAbsentSql(putSql);
    // 无需设置事务管理器
    return store;
}

优缺点 #

• 优点：配置最简单，利用数据库原生特性实现原子性，性能最优；
• 缺点：仅适用于支持ON CONFLICT的数据库（如PostgreSQL），通用性稍弱。

推荐方案 #

优先选择方案三，完全适配PostgreSQL环境，无需额外事务配置即可解决重复处理问题，同时避免与Spring Batch的事务冲突；若需兼容多数据库环境，可选择方案一实现彻底的事务隔离。

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