在Spring Batch中实现跨多数据库的分布式事务

本文旨在指导读者如何在Spring Batch应用中处理涉及多个数据库的分布式事务。当业务需求要求在一个批处理步骤（Step）中同时向不同数据库写入数据时，确保数据一致性至关重要。我们将探讨如何利用CompositeItemWriter聚合多个写入器，并通过配置JtaTransactionManager来协调跨数据库和Spring Batch元数据表的事务，从而实现原子性的数据操作，确保所有写入操作要么全部成功，要么全部回滚。

业务场景概述

在批处理应用中，经常会遇到需要将处理后的数据写入到不同数据库或不同表（可能位于不同数据库实例上）的需求。例如，一个批处理任务可能需要将客户信息写入数据库A的tbl_customer表，同时将订单信息写入数据库B的tbl_order表。在这种情况下，如果其中一个写入操作失败，我们希望所有相关的写入操作都能回滚，以维护数据的一致性。这就引入了分布式事务的需求。

核心策略：组合写入与事务协调

要实现Spring Batch中的分布式事务，核心策略包括两个方面：

组合写入器 (CompositeItemWriter)：用于将数据分发到多个独立的ItemWriter实例，每个实例负责写入一个特定的数据库。JTA分布式事务管理器 (JtaTransactionManager)：用于协调所有参与的数据库（包括业务数据库和Spring Batch的元数据数据库）之间的事务，确保它们作为一个单一的原子操作进行提交或回滚。

1. 配置多数据源与多事务管理器

首先，你需要为每个业务数据库以及Spring Batch的元数据数据库配置独立的DataSource和PlatformTransactionManager。这些事务管理器通常是JdbcTransactionManager或JpaTransactionManager等本地事务管理器。

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.jdbc.datasource.DataSourceTransactionManager;import org.springframework.jdbc.datasource.DriverManagerDataSource;import org.springframework.transaction.PlatformTransactionManager;import javax.sql.DataSource;@Configurationpublic class DataSourceConfig {    // 数据库1 (例如：客户数据)    @Bean    public DataSource customerDataSource() {        DriverManagerDataSource dataSource = new DriverManagerDataSource();        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");        dataSource.setUrl("jdbc:mysql://localhost:3306/db1");        dataSource.setUsername("user1");        dataSource.setPassword("password1");        return dataSource;    }    @Bean    public PlatformTransactionManager customerTransactionManager() {        return new DataSourceTransactionManager(customerDataSource());    }    // 数据库2 (例如：订单数据)    @Bean    public DataSource orderDataSource() {        DriverManagerDataSource dataSource = new DriverManagerDataSource();        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");        dataSource.setUrl("jdbc:mysql://localhost:3306/db2");        dataSource.setUsername("user2");        dataSource.setPassword("password2");        return dataSource;    }    @Bean    public PlatformTransactionManager orderTransactionManager() {        return new DataSourceTransactionManager(orderDataSource());    }    // Spring Batch 元数据数据库    @Bean    public DataSource batchMetaDataDataSource() {        DriverManagerDataSource dataSource = new DriverManagerDataSource();        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");        dataSource.setUrl("jdbc:mysql://localhost:3306/batch_meta");        dataSource.setUsername("batch_user");        dataSource.setPassword("batch_password");        return dataSource;    }    @Bean    public PlatformTransactionManager batchMetaDataTransactionManager() {        return new DataSourceTransactionManager(batchMetaDataDataSource());    }}

2. 配置组合写入器 (CompositeItemWriter)

为每个目标数据库创建一个ItemWriter实例，然后将它们聚合到CompositeItemWriter中。CompositeItemWriter会按顺序调用其委托的ItemWriter。

import org.springframework.batch.item.ItemWriter;import org.springframework.batch.item.support.CompositeItemWriter;import org.springframework.batch.item.database.JdbcBatchItemWriter;import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import javax.sql.DataSource;import java.util.Arrays;import java.util.List;@Configurationpublic class ItemWriterConfig {    // 假设你的数据模型是 Map 或一个POJO    // 这里以 Map 为例    private static class MyItem {        private String customerName;        private String orderId;        // ... other fields        public String getCustomerName() { return customerName; }        public void setCustomerName(String customerName) { this.customerName = customerName; }        public String getOrderId() { return orderId; }        public void setOrderId(String orderId) { this.orderId = orderId; }    }    @Bean    public ItemWriter customerItemWriter(DataSource customerDataSource) {        return new JdbcBatchItemWriterBuilder()                .dataSource(customerDataSource)                .sql("INSERT INTO tbl_customer (name) VALUES (:customerName)")                .beanMapped() // 如果是POJO，使用beanMapped()                .build();    }    @Bean    public ItemWriter orderItemWriter(DataSource orderDataSource) {        return new JdbcBatchItemWriterBuilder()                .dataSource(orderDataSource)                .sql("INSERT INTO tbl_order (order_id) VALUES (:orderId)")                .beanMapped()                .build();    }    @Bean    public CompositeItemWriter compositeItemWriter(            ItemWriter customerItemWriter,            ItemWriter orderItemWriter) {        CompositeItemWriter writer = new CompositeItemWriter();        List<ItemWriter> delegates = Arrays.asList(customerItemWriter, orderItemWriter);        writer.setDelegates(delegates);        return writer;    }}

3. 配置 JTA 分布式事务管理器 (JtaTransactionManager)

JtaTransactionManager是实现分布式事务的关键。它依赖于一个JTA（Java Transaction API）实现，如Atomikos、Narayana或应用服务器（如WildFly、WebLogic）内置的JTA服务。你需要将JTA提供商的UserTransaction和TransactionManager接口的实现注入到JtaTransactionManager中。

腾讯智影-AI数字人

基于AI数字人能力，实现7*24小时AI数字人直播带货，低成本实现直播业务快速增增，全天智能在线直播

73 查看详情

以下以Atomikos为例进行配置：

import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.transaction.jta.JtaTransactionManager;import com.atomikos.icatch.jta.UserTransactionImp;import com.atomikos.icatch.jta.UserTransactionManager;import javax.transaction.SystemException;import javax.transaction.UserTransaction;@Configurationpublic class JtaTransactionManagerConfig {    @Bean(initMethod = "init", destroyMethod = "close")    public UserTransactionManager atomikosTransactionManager() throws SystemException {        UserTransactionManager userTransactionManager = new UserTransactionManager();        userTransactionManager.setForceShutdown(false); // 优雅关闭        return userTransactionManager;    }    @Bean(initMethod = "init", destroyMethod = "close")    public UserTransaction atomikosUserTransaction() throws SystemException {        UserTransactionImp userTransactionImp = new UserTransactionImp();        userTransactionImp.setTransactionTimeout(300); // 事务超时时间，单位秒        return userTransactionImp;    }    @Bean    public JtaTransactionManager jtaTransactionManager(            UserTransaction atomikosUserTransaction,            UserTransactionManager atomikosTransactionManager) {        JtaTransactionManager jtaTm = new JtaTransactionManager();        jtaTm.setUserTransaction(atomikosUserTransaction);        jtaTm.setTransactionManager(atomikosTransactionManager);        // 如果Spring Batch元数据数据库也需要参与JTA事务，        // 确保其DataSource是XA兼容的，并由JTA管理器管理        // 对于Atomikos，通常需要将DataSource配置为AtomikosDataSourceBean        return jtaTm;    }}

重要提示：

XA 数据源： 所有参与分布式事务的DataSource（包括业务数据库和Spring Batch元数据数据库）都必须是XA兼容的。这意味着你需要使用数据库厂商提供的XA驱动，并且将它们配置为XA数据源（例如，使用Atomikos的AtomikosDataSourceBean来包装你的JDBC DataSource）。JTA 提供商： 确保你的项目中引入了JTA提供商的依赖，例如Atomikos或Narayana。

4. 配置 Spring Batch Step

最后，将配置好的JtaTransactionManager注入到你的Spring Batch Step中。这样，该步骤中的所有操作都将在一个由JTA管理器协调的分布式事务中执行。

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.batch.item.support.CompositeItemWriter;import org.springframework.context.annotation.Bean;import org.springframework.context.annotation.Configuration;import org.springframework.transaction.PlatformTransactionManager;import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing; // 导入此注解@Configuration@EnableBatchProcessing // 启用Spring Batch处理public class BatchJobConfig {    // 假设 MyItem 是你的数据模型    private static class MyItem { /* ... */ }    // 假设你已经定义了 ItemReader 和 ItemProcessor    @Bean    public ItemReader myReader() {        // ... 实现你的 ItemReader        return null; // 占位符    }    @Bean    public ItemProcessor myProcessor() {        // ... 实现你的 ItemProcessor        return item -> item; // 简单处理，占位符    }    @Bean    public Step myDistributedTransactionStep(            JobRepository jobRepository,            PlatformTransactionManager jtaTransactionManager, // 注入JTA事务管理器            ItemReader myReader,            ItemProcessor myProcessor,            CompositeItemWriter compositeItemWriter) {        return new StepBuilder("myDistributedTransactionStep", jobRepository)                .chunk(10, jtaTransactionManager) // 将JTA事务管理器传递给chunk方法                .reader(myReader)                .processor(myProcessor)                .writer(compositeItemWriter)                .build();    }    @Bean    public Job myDistributedJob(JobRepository jobRepository, Step myDistributedTransactionStep) {        return new JobBuilder("myDistributedJob", jobRepository)                .start(myDistributedTransactionStep)                .build();    }}

注意事项

JTA 提供商选择： 选择一个可靠的JTA提供商（如Atomikos、Narayana）并正确配置是关键。它们负责管理XA资源和两阶段提交协议。XA 驱动： 确保你的数据库驱动支持XA协议。大多数主流数据库（MySQL, PostgreSQL, Oracle, SQL Server）都提供XA兼容的JDBC驱动。配置复杂性： 分布式事务的配置比本地事务复杂得多，需要仔细配置数据源、事务管理器和JTA提供商。性能考量： 分布式事务引入了额外的开销（如两阶段提交），可能会对批处理的性能产生一定影响。在设计时需要权衡数据一致性与性能。错误处理与回滚： 在分布式事务中，任何一个参与者的失败都将导致整个事务的回滚。Spring Batch的重试和跳过机制仍然有效，但需要确保它们与分布式事务的语义兼容。Spring Batch 元数据： 如果Spring Batch的元数据数据库也需要参与分布式事务（例如，为了确保元数据更新与业务数据更新的原子性），那么其数据源也必须配置为XA兼容，并由JTA管理器协调。

总结

在Spring Batch中实现跨多数据库的分布式事务是一个复杂但必要的任务，尤其是在需要严格数据一致性的企业级应用中。通过合理配置CompositeItemWriter来管理多个数据写入路径，并利用JtaTransactionManager协调底层JTA提供商的分布式事务能力，可以有效地确保批处理操作的原子性。虽然配置过程相对复杂，但它为多数据库环境下的数据完整性提供了强有力的保障。在实施前，务必深入理解JTA规范和所选JTA提供商的特性，并进行充分的测试。

以上就是在Spring Batch中实现跨多数据库的分布式事务的详细内容，更多请关注创想鸟其它相关文章！