title: Spring Batch Multiple Threads
body: <p>I am writing a Spring Batch with idea of scaling it when required.
  My ApplicationContext looks like this </p>
  
  <pre><code>@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&lt;Source&gt; hibernateCursorItemReader = new HibernateCursorItemReader&lt;&gt;();
      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");
  
          }
      };
  }
  </code></pre>
  
  <p>}</p>
  
  <ol>
  <li>With above configuration I am able to read from DB , process the data and write to DB.</li>
  <li>I am using chunk size as 2 and reading 2 records from cursor using
  HibernateCusrsorItem reader and my query to read from DB is based on
  date to pick current date records.</li>
  <li>So far I am able to achieve desired behavior as well as restart
  ability with job only picking records which were not processed 
  due to failure in previous run.</li>
  </ol>
  
  <p>Now my requirement is to make batch use multiple threads to process data and write to DB.</p>
  
  <p>My Processor and writer looks like this</p>
  
  <pre><code>@Component
  public class BatchProcessor implements ItemProcessor&lt;Source,DestinationDto&gt;{
  
  @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&lt;DestinationDto&gt;{
  
  @Autowired
  IBatchDao batchDao;
  
  @Override
  public void write(List&lt;? extends DestinationDto&gt; list) throws Exception {
     System.out.println(Thread.currentThread().getName()+":"+list);
      batchDao.saveToDestination((List&lt;DestinationDto&gt;)list);
  }
  </code></pre>
  
  <p>I updated my step and added a ThreadPoolTaskExecutor as follows</p>
  
  <pre><code>@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();
  
    }
  </code></pre>
  
  <p>After this my processor is getting called by multiple threads but with same source data.
  Is there anything extra i need to do?</p>