Spring Batch 多线程

Question

我正在编写一个 Spring Batch，并在需要时对其进行缩放。 我的 ApplicationContext 看起来像这样

@Configuration
@EnableBatchProcessing
@EnableTransactionManagement
@ComponentScan(basePackages = "in.springbatch")
@PropertySource(value = {"classpath:springbatch.properties"})

public class ApplicationConfig {

@Autowired
Environment environment;

@Autowired
private JobBuilderFactory jobs;

@Autowired
private StepBuilderFactory stepBuilderFactory;

@Bean
public Job job() throws Exception {

    return jobs.get("spring_batch")
            .flow(step()).end()
            .build();
}

@Bean(name = "dataSource", destroyMethod = "close")
public DataSource dataSource() {

    BasicDataSource basicDataSource = new BasicDataSource();



    return basicDataSource;
}

@Bean
public JobRepository jobRepository() throws Exception {
    JobRepositoryFactoryBean jobRepositoryFactoryBean = new JobRepositoryFactoryBean();
    jobRepositoryFactoryBean.setTransactionManager(transactionManager());
    jobRepositoryFactoryBean.setDataSource(dataSource());
    return jobRepositoryFactoryBean.getObject();
}

 @Bean(name = "batchstep")
 public Step step() throws Exception {

    return    stepBuilderFactory.get("batchstep").allowStartIfComplete(true).
    transactionManager(transactionManager()).
          chunk(2).reader(batchReader()).processor(processor()).writer(writer()).build();

  }



@Bean
ItemReader batchReader() throws Exception {
    System.out.println(Thread.currentThread().getName()+"reader");
    HibernateCursorItemReader<Source> hibernateCursorItemReader = new HibernateCursorItemReader<>();
    hibernateCursorItemReader.setQueryString("from Source");
    hibernateCursorItemReader.setFetchSize(2);
    hibernateCursorItemReader.setSessionFactory(sessionFactory().getObject());


    hibernateCursorItemReader.close();
    return hibernateCursorItemReader;
}

@Bean
 public ItemProcessor processor() {
     return new BatchProcessor();
 }

@Bean
public ItemWriter writer() {
    return new BatchWriter();
}

public TaskExecutor taskExecutor(){

    SimpleAsyncTaskExecutor asyncTaskExecutor=new SimpleAsyncTaskExecutor("spring_batch");
    asyncTaskExecutor.setConcurrencyLimit(5);
    return asyncTaskExecutor;


}
@Bean
public LocalSessionFactoryBean sessionFactory() {
    LocalSessionFactoryBean sessionFactory = new LocalSessionFactoryBean();
    sessionFactory.setDataSource(dataSource());
    sessionFactory.setPackagesToScan(new String[]{"in.springbatch.entity"});
    sessionFactory.setHibernateProperties(hibernateProperties());

    return sessionFactory;
}

@Bean
public PersistenceExceptionTranslationPostProcessor exceptionTranslation() {
    return new PersistenceExceptionTranslationPostProcessor();
}

@Bean
@Autowired
public HibernateTransactionManager transactionManager() {
    HibernateTransactionManager txManager = new HibernateTransactionManager();
    txManager.setSessionFactory(sessionFactory().getObject());

    return txManager;
}

Properties hibernateProperties() {
    return new Properties() {
        {
            setProperty("hibernate.hbm2ddl.auto",       environment.getProperty("hibernate.hbm2ddl.auto"));
            setProperty("hibernate.dialect", environment.getProperty("hibernate.dialect"));
            setProperty("hibernate.globally_quoted_identifiers", "false");

        }
    };
}

}

1. 通过上述配置，我可以从 DB 读取、处理数据并写入 DB。
2. 我使用块大小为 2 并使用从游标读取 2 条记录 HibernateCursorItem 阅读器和我从数据库中读取的查询是基于 date 选择当前日期记录。
3. 到目前为止，我能够实现所需的行为以及重新启动 仅处理未处理的作业拣货记录的能力 由于之前的运行失败。

现在我的要求是批量使用多个线程来处理数据并写入数据库。

我的处理器和写入器看起来像这样

@Component
public class BatchProcessor implements ItemProcessor<Source,DestinationDto>{

@Override
public DestinationDto process(Source source) throws Exception {

        System.out.println(Thread.currentThread().getName()+":"+source);
        DestinationDto destination=new DestinationDto();
        destination.setName(source.getName());
        destination.setValue(source.getValue());
        destination.setSourceId(source.getSourceId().toString());

    return destination;
}
@Component
public class BatchWriter implements ItemWriter<DestinationDto>{

@Autowired
IBatchDao batchDao;

@Override
public void write(List<? extends DestinationDto> list) throws Exception {
   System.out.println(Thread.currentThread().getName()+":"+list);
    batchDao.saveToDestination((List<DestinationDto>)list);
}

我更新了我的步骤并添加了一个 ThreadPoolTask​​Executor 如下

@Bean(name = "batchstep")
public Step step() throws Exception {

    return  stepBuilderFactory.get("batchstep").allowStartIfComplete(true).
     transactionManager(transactionManager()).chunk(1).reader(batchReader()).
     processor(processor()).writer(writer()).taskExecutor(taskExecutor()).build();

  }

在此之后，我的处理器被多个线程调用但具有相同的源数据。 还有什么需要我做的吗？

Answer 1

这是个大问题

1. 最好的办法是查看 Spring Batch 文档 (Here) 中的 Scaling and Parallel Processing 一章

2. spring 批处理示例中可能存在一些多线程示例 (Here)

3. 线程化 Spring 批处理作业的一种简单方法是创建一个未来处理器 - 您将所有处理逻辑放在一个未来对象中，而您的 spring-processor 类只将对象添加到未来。您的作家类然后等待未来完成，然后再执行写入过程。抱歉，我也没有样本可以为您指出这一点 - 但如果您有具体问题，我可以尝试回答！