Multithreading with StoredProcedureItemReader

Viewed 26

Is it possible to have multiple threads with StoredProcedureItemReader? I have done multithreading with PageReader but not sure if it will work with StoredProcedureItemReader.

Below is the job configuration for using a StoredProcedureReader. I have wrapped the reader in Thread safe reader. I want to use ThreadPoolTaskExecutor but not able to figure out how I can do partition for each thread with the Stored procedure.

***@Configuration
public class SpPocJobConfigurationMT {
    private DataSource dataSource;
    /**
     * The Job builder factory.
     */
    private JobBuilderFactory jobBuilderFactory;
    /**
     * The Jdbc template.
     */
    @Autowired
    JdbcTemplate jdbcTemplate;
    /**
     * The Step builder factory.
     */
    private StepBuilderFactory stepBuilderFactory;
    @Autowired
    private BillingRecordAuditRepository billingRecordAuditRepository;
    @Autowired
    private StagingMortgageDataTxnRepository stagingMortgageDataTxnRepository;
    private SystemRepository systemRepository;
    @Autowired
    public SpPocJobConfigurationMT(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory, SystemRepository systemRepository, DataSource dataSource) {
        Assert.notNull(systemRepository, "SystemRepository cannot be null");
        Assert.notNull(jobBuilderFactory, "JobBuilderFactory cannot be null");
        Assert.notNull(stepBuilderFactory, "StepBuilderFactory cannot be null");
        Assert.notNull(dataSource, "DataSource cannot be null");
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
        this.systemRepository = systemRepository;
        this.dataSource = dataSource;
    }
   
    @Bean
    @Transactional
    @Description(value = "")
    public Job SpPocJobMT() throws Exception {
        return jobBuilderFactory.get("spPocJobMT")
                .start(spPocStepMT())
                .build();
    }
   
    @Bean
    public Step spPocStepMT() throws Exception {
        return stepBuilderFactory.get("spPocStepMT")
                .allowStartIfComplete(false)
                .<StagingDataDto,StagingDataDto> chunk(20)
                .reader(sybcSpReaderMT())
                .processor(spPocProcessorMT())
                .writer(spPocWriterMT())
              //  .taskExecutor(new ThreadPoolTaskExecutor ())
             //   .taskExecutor(new SimpleAsyncTaskExecutor())
                .build();
    }
   
    @Bean
    public SpPocWriter spPocWriterMT() {
        return new SpPocWriter(this.billingRepository, this.stagingTxnRepository);
    }
    @Bean
    public SpPocProcessor spPocProcessorMT() {
        return new SpPocProcessor();
    }
    @Bean
    @StepScope
    public SynchronizedItemStreamReader sybcSpReaderMT() {
        StoredProcedureItemReader reader = new StoredProcedureItemReader();
        SqlParameter[] parameters = {new SqlParameter("@p_id", OracleTypes.NUMBER)
                , new SqlOutParameter("@p_out_c1", OracleTypes.CURSOR)
                  , new SqlOutParameter("@p_out_c2", OracleTypes.CURSOR)
        };
        reader.setDataSource(dataSource);
        reader.setProcedureName("SP_POC_FINAL");
        reader.setRowMapper(new SPRowMapper());
        reader.setRefCursorPosition(3);
        reader.setPreparedStatementSetter(new MyItemPreparedStatementSetter());
        reader.setParameters(parameters);
        reader.setSaveState(false);
        reader.setVerifyCursorPosition(false);
        SynchronizedItemStreamReader synchronizedItemStreamReader = new SynchronizedItemStreamReader();
        synchronizedItemStreamReader.setDelegate(reader);
        return synchronizedItemStreamReader;
    }
    public class MyItemPreparedStatementSetter implements PreparedStatementSetter {
        @Override
        public void setValues(PreparedStatement ps) throws SQLException {
            ps.setInt(1, 1);
            ((CallableStatement) ps).registerOutParameter(2, OracleTypes.CURSOR);
              ((CallableStatement) ps).registerOutParameter(3, OracleTypes.CURSOR);
        }
    }
}***
1 Answers

The StoredProcedureItemReader extends AbstractItemCountingItemStreamItemReader which is not thread-safe, please check its javadoc.

If you want to use the StoredProcedureItemReader in a multi-threaded step, you need to wrap it in a SynchronizedItemStreamReader or make it step-scoped.

Related