Spring Batch 中实现跨数据库分布式事务的策略

import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.item.support.CompositeItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.Arrays;
import java.util.List;

@Configuration
public class ItemWriterConfig {

    // 假设您已经定义了db1和db2的JdbcBatchItemWriter
    // 例如：db1CustomerWriter 用于写入db1的tbl_customer
    // db2OrderWriter 用于写入db2的tbl_order

    @Bean
    public ItemWriter<YourItemType> db1CustomerWriter() {
        // 配置写入db1的ItemWriter，例如JdbcBatchItemWriter
        // ...
        return null; // 实际应返回配置好的writer实例
    }

    @Bean
    public ItemWriter<YourItemType> db2OrderWriter() {
        // 配置写入db2的ItemWriter，例如JdbcBatchItemWriter
        // ...
        return null; // 实际应返回配置好的writer实例
    }

    @Bean
    public CompositeItemWriter<YourItemType> compositeDbItemWriter(
            ItemWriter<YourItemType> db1CustomerWriter,
            ItemWriter<YourItemType> db2OrderWriter) {

        CompositeItemWriter<YourItemType> writer = new CompositeItemWriter<>();
        // 将所有目标数据库的ItemWriter作为委托添加到CompositeItemWriter中
        List<ItemWriter<? super YourItemType>> delegates = Arrays.asList(db1CustomerWriter, db2OrderWriter);
        writer.setDelegates(delegates);
        return writer;
    }
}

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.jta.JtaTransactionManager;

@Configuration
public class TransactionManagerConfig {

    @Bean
    public PlatformTransactionManager jtaTransactionManager() {
        // JtaTransactionManager 会自动查找并使用JTA环境中的UserTransaction和TransactionManager。
        // 如果在应用服务器中运行，通常无需额外配置。
        // 如果使用独立的JTA提供者（如Atomikos），可能需要进一步配置其UserTransactionManager和UserTransactionImp。
        JtaTransactionManager jtaTm = new JtaTransactionManager();
        // 如果需要，可以指定JNDI名称来查找JTA服务
        // jtaTm.setUserTransactionName("java:comp/UserTransaction");
        // jtaTm.setTransactionManagerName("java:comp/TransactionManager");
        return jtaTm;
    }
}

import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.job.builder.JobBuilder;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemWriter;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.PlatformTransactionManager;

@Configuration
public class BatchJobConfig {

    // 假设您已经定义了ItemReader和ItemProcessor
    @Bean
    public ItemReader<YourItemType> itemReader() {
        // ...
        return null;
    }

    @Bean
    public ItemProcessor<YourItemType, YourItemType> itemProcessor() {
        // ...
        return null;
    }

    @Bean
    public Step myDistributedTransactionStep(
            JobRepository jobRepository,
            PlatformTransactionManager jtaTransactionManager, // 注入JTA事务管理器
            ItemReader<YourItemType> itemReader,
            CompositeItemWriter<YourItemType> compositeDbItemWriter, // 注入复合写入器
            ItemProcessor<YourItemType, YourItemType> itemProcessor) {

        return new StepBuilder("myDistributedTransactionStep", jobRepository)
                .<YourItemType, YourItemType>chunk(10) // 批处理大小
                .reader(itemReader)
                .processor(itemProcessor)
                .writer(compositeDbItemWriter) // 使用复合写入器
                .transactionManager(jtaTransactionManager) // 将JTA事务管理器应用于步骤
                .build();
    }

    @Bean
    public Job myDistributedJob(JobRepository jobRepository, Step myDistributedTransactionStep) {
        return new JobBuilder("myDistributedJob", jobRepository)
                .start(myDistributedTransactionStep)
                .build();
    }
}