Spring Batch双数据源配置问题：Repository读写器与JPA默认行为处理

1. 定义双数据源Bean #

先在配置类中分别定义Spring Batch元数据专用数据源和业务读写专用数据源，通过@Primary标记Spring Batch的数据源，避免框架自动选择其他数据源：

@Configuration
public class DataSourceConfig {

    // Spring Batch元数据数据源（替换为你的实际配置）
    @Bean(name = "batchDataSource")
    @Primary
    public DataSource batchDataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        dataSource.setUrl("jdbc:mysql://localhost:3306/batch_metadata");
        dataSource.setUsername("root");
        dataSource.setPassword("password");
        return dataSource;
    }

    // 业务读写数据源（替换为你的另一个数据库配置）
    @Bean(name = "businessDataSource")
    public DataSource businessDataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName("org.postgresql.Driver"); // 示例为PostgreSQL，按需调整
        dataSource.setUrl("jdbc:postgresql://localhost:5432/business_db");
        dataSource.setUsername("postgres");
        dataSource.setPassword("postgres");
        return dataSource;
    }
}

2. 强制Spring Batch使用指定数据源 #

通过自定义BatchConfigurer覆盖默认配置，明确指定Spring Batch内部操作（元数据读写）使用的数据源：

@Configuration
@EnableBatchProcessing
public class BatchConfig implements BatchConfigurer {

    private final DataSource batchDataSource;

    public BatchConfig(@Qualifier("batchDataSource") DataSource batchDataSource) {
        this.batchDataSource = batchDataSource;
    }

    @Override
    public JobRepository getJobRepository() throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        factory.setDataSource(batchDataSource);
        factory.setTransactionManager(getTransactionManager());
        factory.setDatabaseType(DatabaseType.MYSQL.name()); // 匹配元数据库类型
        factory.setTablePrefix("BATCH_"); // 对应Spring Batch元数据表前缀
        factory.afterPropertiesSet();
        return factory.getObject();
    }

    @Override
    public PlatformTransactionManager getTransactionManager() throws Exception {
        return new DataSourceTransactionManager(batchDataSource);
    }

    @Override
    public JobLauncher getJobLauncher() throws Exception {
        SimpleJobLauncher launcher = new SimpleJobLauncher();
        launcher.setJobRepository(getJobRepository());
        launcher.afterPropertiesSet();
        return launcher;
    }

    @Override
    public JobExplorer getJobExplorer() throws Exception {
        JobExplorerFactoryBean factory = new JobExplorerFactoryBean();
        factory.setDataSource(batchDataSource);
        factory.setTablePrefix("BATCH_");
        factory.afterPropertiesSet();
        return factory.getObject();
    }
}

3. 绑定业务Repository到业务数据源 #

配置独立的JPA组件，让业务Repository关联到业务数据源：

@Configuration
@EnableJpaRepositories(
        basePackages = "com.yourpackage.business.repository", // 业务Repository所在包
        entityManagerFactoryRef = "businessEntityManagerFactory",
        transactionManagerRef = "businessTransactionManager"
)
public class BusinessJpaConfig {

    @Autowired
    @Qualifier("businessDataSource")
    private DataSource businessDataSource;

    @Bean(name = "businessEntityManagerFactory")
    public LocalContainerEntityManagerFactoryBean businessEntityManagerFactory(EntityManagerFactoryBuilder builder) {
        return builder
                .dataSource(businessDataSource)
                .packages("com.yourpackage.business.entity") // 业务实体类所在包
                .persistenceUnit("businessPU")
                .build();
    }

    @Bean(name = "businessTransactionManager")
    public PlatformTransactionManager businessTransactionManager(
            @Qualifier("businessEntityManagerFactory") EntityManagerFactory entityManagerFactory) {
        return new JpaTransactionManager(entityManagerFactory);
    }
}

4. 使用RepositoryItemReader/Writer实现业务读写 #

直接注入业务Repository即可，Step中指定业务事务管理器确保操作在业务数据源事务下执行：

@Configuration
public class JobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private BusinessRepository businessRepository;

    @Autowired
    private ProcessedEntityRepository processedEntityRepository;

    @Autowired
    @Qualifier("businessTransactionManager")
    private PlatformTransactionManager businessTransactionManager;

    @Bean
    public Step dataProcessingStep() {
        return stepBuilderFactory.get("dataProcessingStep")
                .<BusinessEntity, ProcessedEntity>chunk(100)
                .reader(repositoryItemReader())
                .processor(dataProcessor())
                .writer(repositoryItemWriter())
                .transactionManager(businessTransactionManager)
                .build();
    }

    @Bean
    public RepositoryItemReader<BusinessEntity> repositoryItemReader() {
        RepositoryItemReader<BusinessEntity> reader = new RepositoryItemReader<>();
        reader.setRepository(businessRepository);
        reader.setMethodName("findByStatus"); // 自定义查询方法，按需调整
        reader.setPageSize(100);
        // 设置排序规则，避免分页读取数据混乱
        Map<String, Sort.Direction> sortMap = new HashMap<>();
        sortMap.put("id", Sort.Direction.ASC);
        reader.setSort(sortMap);
        return reader;
    }

    @Bean
    public RepositoryItemWriter<ProcessedEntity> repositoryItemWriter() {
        RepositoryItemWriter<ProcessedEntity> writer = new RepositoryItemWriter<>();
        writer.setRepository(processedEntityRepository);
        writer.setMethodName("saveAll"); // 批量保存方法
        return writer;
    }

    @Bean
    public ItemProcessor<BusinessEntity, ProcessedEntity> dataProcessor() {
        return item -> {
            ProcessedEntity processed = new ProcessedEntity();
            processed.setContent(item.getContent());
            processed.setProcessedTime(LocalDateTime.now());
            return processed;
        };
    }
}

核心注意事项 #

• 严格区分元数据数据源和业务数据源，避免事务或连接池冲突
• 通过BatchConfigurer彻底隔离Spring Batch内部操作与业务数据源
• 业务Repository必须绑定到独立的JPA配置，确保读写操作指向目标数据库

内容的提问来源于stack exchange，提问作者Buddhima Chamaranga