Spring Batch 中多个数据源的使用

Question

我正在尝试在 Spring Batch 中配置几个数据源。在启动时，Spring Batch 抛出以下异常：

To use the default BatchConfigurer the context must contain no more thanone DataSource, found 2

批处理配置中的片段

@Configuration
@EnableBatchProcessing 
public class BatchJobConfiguration {

    @Primary
    @Bean(name = "baseDatasource")
    public DataSource dataSource() {
         // first datasource definition here
    }
    @Bean(name = "secondaryDataSource")
    public DataSource dataSource2() {
         // second datasource definition here
    }
    ...
}

不知道为什么我会看到这个异常，因为我已经看到了一些基于 xml 的 Spring 批处理配置，它声明了多个数据源。我正在使用 Spring Batch 核心版本 3.0.1.RELEASE 和 Spring Boot 版本 1.1.5.RELEASE。任何帮助将不胜感激。

Answer 1

您可以在下面定义 bean 并确保您的 application.properties 文件具有所需的条目

@Configuration
@PropertySource("classpath:application.properties")
public class DataSourceConfig {

    @Primary
    @Bean(name = "abcDataSource")
    @ConfigurationProperties(prefix = "abc.datasource")
    public DataSource dataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }


    @Bean(name = "xyzDataSource")
    @ConfigurationProperties(prefix = "xyz.datasource")
    public DataSource xyzDataSource() {
        return DataSourceBuilder.create().type(HikariDataSource.class).build();
    }
}

application.properties

abc.datasource.jdbc-url=XXXXX
abc.datasource.username=XXXXX
abc.datasource.password=xxxxx
abc.datasource.driver-class-name=org.postgresql.Driver

...........
...........
...........
...........

这里可以参考：Spring Boot Configure and Use Two DataSources

Answer 2

最简单的解决方案是扩展 DefaultBatchConfigurer 并通过限定符自动连接您的数据源：

@Component
public class MyBatchConfigurer extends DefaultBatchConfigurer {

    /**
     * Initialize the BatchConfigurer to use the datasource of your choosing
     * @param firstDataSource
     */
    @Autowired
    public MyBatchConfigurer(@Qualifier("firstDataSource") DataSource firstDataSource) {
        super(firstDataSource);
    }
}

旁注（因为这也涉及使用多个数据源）：如果您使用自动配置来运行数据初始化脚本，您可能会注意到它没有在您期望的数据源上进行初始化。对于这个问题，看看这个：https://github.com/spring-projects/spring-boot/issues/9528

Answer 3

我想在这里提供一个解决方案，这与@vanarchi 回答的解决方案非常相似，但我设法将所有必要的配置放入一个类中。

为了完整起见，这里的解决方案假设主数据源是hsql。

@Configuration
@EnableBatchProcessing
public class BatchConfiguration extends DefaultBatchConfigurer {

@Bean
@Primary
public DataSource batchDataSource() {

    // no need shutdown, EmbeddedDatabaseFactoryBean will take care of this
    EmbeddedDatabaseBuilder builder = new EmbeddedDatabaseBuilder();
    EmbeddedDatabase embeddedDatabase = builder
            .addScript("classpath:org/springframework/batch/core/schema-drop-hsqldb.sql")
            .addScript("classpath:org/springframework/batch/core/schema-hsqldb.sql")
            .setType(EmbeddedDatabaseType.HSQL) //.H2 or .DERBY
            .build();
    return embeddedDatabase;
}

@Override
protected JobRepository createJobRepository() throws Exception {
    JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
    factory.setDataSource(batchDataSource());
    factory.setTransactionManager(transactionManager());
    factory.afterPropertiesSet();

    return (JobRepository) factory.getObject();
}

private ResourcelessTransactionManager transactionManager() {
    return new ResourcelessTransactionManager();
}

//NOTE: the code below is just to provide developer an easy way to access the in-momery hsql datasource, as we configured it to the primary datasource to store batch job related data. Default username : sa, password : ''
@PostConstruct
public void getDbManager(){
    DatabaseManagerSwing.main(
            new String[] { "--url", "jdbc:hsqldb:mem:testdb", "--user", "sa", "--password", ""});
}

}

这个解决方案的三个关键点：

1. 该类使用@EnableBatchProcessing 和@Configuration 进行注释，并从DefaultBatchConfigurer 扩展而来。通过这样做，我们指示 spring-batch 在 AbstractBatchConfiguration 尝试查找 BatchConfigurer 时使用我们自定义的批处理配置器；
2. 将batchDataSource bean注解为@Primary，指示spring-batch使用该数据源作为其存储9个作业相关表的数据源。
3. 重写 protected JobRepository createJobRepository() throws Exception 方法，这使得 jobRepository bean 使用主数据源，并使用与其他数据源不同的 transactionManager 实例。

Answer 4

如果我可以补充上述问题，那么每个 DS 都有 2 个事务上下文的含义。如何将 XA 事务与批处理步骤集成，因为我们需要确保步骤级别的 TXN 管理？要求就像在批处理步骤中我们需要以下内容。

1. 从 DS 1 读取 -- jpaItemReader
2. 写入 DS2 - JPAItemwriter
3. 从 DS2 读取 - JPAItemreader
4. 写入 Ds1 - JPAItemwriter
5. 提交所有 txns 步骤完成。

Answer 5

AbstractBatchConfiguration 首先尝试在容器中查找 BatchConfigurer，如果找不到，则尝试自行创建它 - 这是在容器中有多个 DataSource bean 时抛出 IllegalStateException 的地方。

解决问题的方法是防止在AbstractBatchConfiguration 中创建DefaultBatchConfigurer bean。 为此，我们提示通过 Spring 容器使用 @Component 注释创建 DefaultBatchConfigurer：

@EnableBatchProcessing 所在的配置类我们可以用@ComponentScan 进行注解，以扫描包含从DefaultBatchConfigurer 派生的空类的包：

package batch_config;
...
@EnableBatchProcessing
@ComponentScan(basePackageClasses = MyBatchConfigurer.class)
public class MyBatchConfig {
    ...
}

空派生类的完整代码在这里：

package batch_config.components;
import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
import org.springframework.stereotype.Component;
@Component
public class MyBatchConfigurer extends DefaultBatchConfigurer {
}

在此配置中，@Primary 注释适用于 DataSource bean，如下例所示：

@Configuration
public class BatchTestDatabaseConfig {
    @Bean
    @Primary
    public DataSource dataSource()
    {
        return .........;
    }
}

这适用于 Spring Batch 版本 3.0.3.RELEASE

使DataSource 上的@Primary 注释起作用的最简单解决方案可能只是添加@ComponentScan(basePackageClasses = DefaultBatchConfigurer.class) 和@EnableBatchProcessing 注释：

@Configuration
@EnableBatchProcessing
@ComponentScan(basePackageClasses = DefaultBatchConfigurer.class)
public class MyBatchConfig {

Answer 6

您必须提供自己的 BatchConfigurer。 Spring 不想为你做那个决定

@Configuration
@EnableBatchProcessing
public class BatchConfig {

     @Bean
      BatchConfigurer configurer(@Qualifier("batchDataSource") DataSource dataSource){
        return new DefaultBatchConfigurer(dataSource);
      }

...

Answer 7

首先，创建一个自定义的 BatchConfigurer

@Configuration
@Component
public class TwoDataSourcesBatchConfigurer implements BatchConfigurer {

    @Autowired
    @Qualifier("dataSource1")
    DataSource dataSource;

    @Override
    public JobExplorer getJobExplorer() throws Exception {
        ...
    }

    @Override
    public JobLauncher getJobLauncher() throws Exception {
        ...
    }

    @Override
    public JobRepository getJobRepository() throws Exception {
        JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
        // use the autowired data source
        factory.setDataSource(dataSource);
        factory.setTransactionManager(getTransactionManager());
        factory.afterPropertiesSet();
        return factory.getObject();
    }

    @Override
    public PlatformTransactionManager getTransactionManager() throws Exception                      {
        ...
    }

}

那么，

@Configuration
@EnableBatchProcessing
@ComponentScan("package")
public class JobConfig {
    // define job, step, ...
}