业务场景概述
在批处理应用中，经常会遇到需要将处理后的数据写入到不同数据库或不同表（可能位于不同数据库实例上）的需求。例如，一个批处理任务可能需要将客户信息写入数据库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;
@Configuration
public 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;
@Configuration
public 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> 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中。
华友协同办公自动化OA系统
下载
华友协同办公管理系统(华友OA)，基于微软最新的.net 2.0平台和SQL Server数据库，集成强大的Ajax技术，采用多层分布式架构，实现统一办公平台，功能强大、价格便宜，是适用于企事业单位的通用型网络协同办公系统。 系统秉承协同办公的思想，集成即时通讯、日记管理、通知管理、邮件管理、新闻、考勤管理、短信管理、个人文件柜、日程安排、工作计划、工作日清、通讯录、公文流转、论坛、在线调查、
以下以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;
@Configuration
public 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提供商的特性，并进行充分的测试。
