spring事务跨线程 spring事务可以跨多线程吗

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

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

首先，你需要为每个业务数据库以及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());    }}

为每个目标数据库创建一个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<String, Object> 或一个POJO    // 这里以 Map<String, Object> 为例    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<MyItem> customerItemWriter(DataSource customerDataSource) {        return new JdbcBatchItemWriterBuilder<MyItem>()                .dataSource(customerDataSource)                .sql("INSERT INTO tbl_customer (name) VALUES (:customerName)")                .beanMapped() // 如果是POJO，使用beanMapped()                .build();    }    @Bean    public ItemWriter<MyItem> orderItemWriter(DataSource orderDataSource) {        return new JdbcBatchItemWriterBuilder<MyItem>()                .dataSource(orderDataSource)                .sql("INSERT INTO tbl_order (order_id) VALUES (:orderId)")                .beanMapped()                .build();    }    @Bean    public CompositeItemWriter<MyItem> compositeItemWriter(            ItemWriter<MyItem> customerItemWriter,            ItemWriter<MyItem> orderItemWriter) {        CompositeItemWriter<MyItem> writer = new CompositeItemWriter<>();        List<ItemWriter<? super MyItem>> delegates = Arrays.asList(customerItemWriter, orderItemWriter);        writer.setDelegates(delegates);        return writer;    }}

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

以下以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;    }}

重要提示：

最后，将配置好的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<MyItem> myReader() {        // ... 实现你的 ItemReader        return null; // 占位符    }    @Bean    public ItemProcessor<MyItem, MyItem> myProcessor() {        // ... 实现你的 ItemProcessor        return item -> item; // 简单处理，占位符    }    @Bean    public Step myDistributedTransactionStep(            JobRepository jobRepository,            PlatformTransactionManager jtaTransactionManager, // 注入JTA事务管理器            ItemReader<MyItem> myReader,            ItemProcessor<MyItem, MyItem> myProcessor,            CompositeItemWriter<MyItem> compositeItemWriter) {        return new StepBuilder("myDistributedTransactionStep", jobRepository)                .<MyItem, MyItem>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();    }}

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

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