JdbcCursorItemReader的分區程序 - 讀取程序在讀取之前必須打開

我有一個工作的彈簧批處理作業，當我嘗試使用分區程序創建多線程時，我開始讓Reader必須打開才能讀取。JdbcCursorItemReader的分區程序 - 讀取程序在讀取之前必須打開

org.springframework.batch.item.ReaderNotOpenException: Reader must be open before it can be read. 
    at org.springframework.batch.item.database.AbstractCursorItemReader.doRead(AbstractCursorItemReader.java:443) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader.read(AbstractItemCountingItemStreamItemReader.java:88) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101] 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101] 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101] 
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101] 
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:302) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:133) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:121) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:208) ~[spring-aop-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at com.sun.proxy.$Proxy58.read(Unknown Source) ~[na:na] 
    at org.springframework.batch.core.step.item.SimpleChunkProvider.doRead(SimpleChunkProvider.java:91) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.item.SimpleChunkProvider.read(SimpleChunkProvider.java:157) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.item.SimpleChunkProvider$1.doInIteration(SimpleChunkProvider.java:116) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.item.SimpleChunkProvider.provide(SimpleChunkProvider.java:110) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:69) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:406) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:330) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133) ~[spring-tx-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:81) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144) ~[spring-batch-infrastructure-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:139) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:136) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_101] 
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101] 

2016-08-01 14:26:14.700 ERROR 12716 --- [   main] o.s.batch.core.step.AbstractStep   : Encountered an error executing step partitionStep in job exportMasterListCsv 

org.springframework.batch.core.JobExecutionException: Partition handler returned an unsuccessful step 
    at org.springframework.batch.core.partition.support.PartitionStep.doExecute(PartitionStep.java:112) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200) ~[spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.job.SimpleStepHandler.handleStep(SimpleStepHandler.java:148) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.job.AbstractJob.handleStep(AbstractJob.java:392) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.job.SimpleJob.doExecute(SimpleJob.java:135) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.job.AbstractJob.execute(AbstractJob.java:306) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.core.launch.support.SimpleJobLauncher$1.run(SimpleJobLauncher.java:135) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50) [spring-core-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.batch.core.launch.support.SimpleJobLauncher.run(SimpleJobLauncher.java:128) [spring-batch-core-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at org.springframework.batch.test.JobLauncherTestUtils.launchJob(JobLauncherTestUtils.java:152) [spring-batch-test-3.0.7.RELEASE.jar:3.0.7.RELEASE] 
    at com.myer.pricing.onlinestore.export.job.ExportMasterListCsvTest.testLaunchJob(ExportMasterListCsvTest.java:60) [test-classes/:na] 
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:1.8.0_101] 
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[na:1.8.0_101] 
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:1.8.0_101] 
    at java.lang.reflect.Method.invoke(Method.java:498) ~[na:1.8.0_101] 
    at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:47) [junit-4.11.jar:na] 
    at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12) [junit-4.11.jar:na] 
    at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:44) [junit-4.11.jar:na] 
    at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17) [junit-4.11.jar:na] 
    at org.junit.internal.runners.statements.RunBefores.evaluate(RunBefores.java:26) [junit-4.11.jar:na] 
    at org.springframework.test.context.junit4.statements.RunBeforeTestMethodCallbacks.evaluate(RunBeforeTestMethodCallbacks.java:75) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.test.context.junit4.statements.RunAfterTestMethodCallbacks.evaluate(RunAfterTestMethodCallbacks.java:86) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.test.context.junit4.statements.SpringRepeat.evaluate(SpringRepeat.java:84) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.junit.runners.ParentRunner.runLeaf(ParentRunner.java:271) [junit-4.11.jar:na] 
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:254) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.runChild(SpringJUnit4ClassRunner.java:89) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.junit.runners.ParentRunner$3.run(ParentRunner.java:238) [junit-4.11.jar:na] 
    at org.junit.runners.ParentRunner$1.schedule(ParentRunner.java:63) [junit-4.11.jar:na] 
    at org.junit.runners.ParentRunner.runChildren(ParentRunner.java:236) [junit-4.11.jar:na] 
    at org.junit.runners.ParentRunner.access$000(ParentRunner.java:53) [junit-4.11.jar:na] 
    at org.junit.runners.ParentRunner$2.evaluate(ParentRunner.java:229) [junit-4.11.jar:na] 
    at org.springframework.test.context.junit4.statements.RunBeforeTestClassCallbacks.evaluate(RunBeforeTestClassCallbacks.java:61) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.springframework.test.context.junit4.statements.RunAfterTestClassCallbacks.evaluate(RunAfterTestClassCallbacks.java:70) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.junit.runners.ParentRunner.run(ParentRunner.java:309) [junit-4.11.jar:na] 
    at org.springframework.test.context.junit4.SpringJUnit4ClassRunner.run(SpringJUnit4ClassRunner.java:193) [spring-test-4.2.6.RELEASE.jar:4.2.6.RELEASE] 
    at org.eclipse.jdt.internal.junit4.runner.JUnit4TestReference.run(JUnit4TestReference.java:86) [.cp/:na] 
    at org.eclipse.jdt.internal.junit.runner.TestExecution.run(TestExecution.java:38) [.cp/:na] 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:459) [.cp/:na] 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.runTests(RemoteTestRunner.java:678) [.cp/:na] 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.run(RemoteTestRunner.java:382) [.cp/:na] 
    at org.eclipse.jdt.internal.junit.runner.RemoteTestRunner.main(RemoteTestRunner.java:192) [.cp/:na]

我只是想知道如果JdbcCursorItemReader不是線程安全的，因此也許爲什麼我得到這個錯誤？

這裏是大致我如何設定我的工作了......

@Configuration 
public class ExportMasterListCsvJobConfig { 

public static final String JOB_NAME = "exportMasterListCsv"; 

@Autowired 
public JobBuilderFactory jobBuilderFactory; 

@Autowired 
public StepBuilderFactory stepBuilderFactory; 

@Bean 
public TaskExecutor taskExecutor() { 
    return new SimpleAsyncTaskExecutor(); 
} 


/** 
* Create job to export the master list from the online pricing staging db. 
* 
* @param readStgDbAndExportMasterListStep read online staging db step and export to csv 
* @return job to export the master list from the online pricing staging db. 
*/ 
@Bean 
public Job exportMasterListCsvJob(@Qualifier("partitionStep") Step partitionStep) { 

    return jobBuilderFactory.get(JOB_NAME) 
      .incrementer(new RunIdIncrementer()) 
      .start(partitionStep) 
      .build(); 
} 

    @Bean 
    public Step partitionStep(@Qualifier("readStgDbAndExportMasterListStep") Step readStgDbAndExportMasterListStep){ 
    return stepBuilderFactory.get("partitionStep") 
     .partitioner(readStgDbAndExportMasterListStep) 
     .partitioner("readStgDbAndExportMasterListStep", partitioner()) 
     .taskExecutor(taskExecutor()) 
     .build(); 
    } 

/** 
* Read from online pricing staging db. 
* 
* @param chunkSize 
* @param queryOnlineStagingDbReaderForMasterList 
* @param masterListOutputProcessor 
* @param masterListFileWriter 
* @return step to read from online pricing staging db. 
*/ 
@Bean 
public Step readStgDbAndExportMasterListStep(
     @Value("${exportMasterListCsv.generateMasterListRows.chunkSize}") int chunkSize, 
     @Qualifier("queryOnlineStagingDbReaderForMasterList") ItemReader<MasterList> queryOnlineStagingDbReaderForMasterList, 
     @Qualifier("masterListOutputProcessor") ItemProcessor<MasterList,MasterList> masterListOutputProcessor, 
     @Qualifier("masterListFileWriter") ItemWriter<MasterList> masterListFileWriter) { 

    return stepBuilderFactory.get("readStgDbAndExportMasterListStep") 
       .<MasterList,MasterList>chunk(chunkSize) 
       .reader(queryOnlineStagingDbReaderForMasterList) 
       .processor(masterListOutputProcessor) 
       .writer(masterListFileWriter) 
       .build(); 

} 

/** 
* Query and map rows from online pricing staging db into a master list object. 
* 
* @param onlineStagingDb 
* @param masterListSql 
* @return 
*/ 
@Bean 
@StepScope 
public ItemReader<MasterList> queryOnlineStagingDbReaderForMasterList(
     DataSource onlineStagingDb, 
     @Value("#{stepExecutionContext[divisionId]}") Integer divisionId, 
     @Value("${exportMasterListCsv.generateMasterListRows.masterListSql}") String masterListSql) { 

    JdbcCursorItemReader<MasterList> reader = new JdbcCursorItemReader<>(); 

    reader.setDataSource(onlineStagingDb); 
    reader.setSql(masterListSql); 
    reader.setPreparedStatementSetter(new PreparedStatementSetter() { 
     public void setValues(PreparedStatement ps) throws SQLException { 
      ps.setInt(1, divisionId); 
     } 
    }); 


    reader.setRowMapper(new RowMapper<MasterList>() { 
     @Override 
     public MasterList mapRow(ResultSet resultSet, int i) throws SQLException { 
      MasterList masterList = new MasterList(); 
      masterList.setL1(resultSet.getString(DaoConstants.COLUMN_HEADER_LEVEL_ONE)); 

      return masterList; 
     } 
    });   
    return reader; 
} 

@Bean 
public Partitioner partitioner(){ 
    return new DivisionPartitioner(); 
}

我一直在做的淨一些閱讀。這可能與https://jira.spring.io/browse/BATCH-1301有關嗎？感謝

來源

2016-07-31 Richie

只有部分代碼被正確粘貼; 'queryOnlineStagingDbReaderForMasterList'被切斷。 – Brad

添加了一些我的實現爲你看看 – Richie

我只是想知道如果JdbcCursorItemReader不是線程安全的，因此也許爲什麼我收到這個錯誤？

JdbcCursorItemReader是一個AbstractCursorItemReader，它是一個AbstractItemCountingItemStreamItemReader。

按照AbstractItemCountingItemStreamItemReader的文檔的實現是不線程

http://docs.spring.io/spring-batch/apidocs/org/springframework/batch/item/support/AbstractItemCountingItemStreamItemReader.html

爲ItemReaders抽象超通過存儲項目在執行上下文（計數支持重啓因此需要項目排序到在運行之間保存）。 子類本質上不是線程安全的。

向線程問題日誌提示的這一部分：

at java.util.concurrent.FutureTask.run(FutureTask.java:266) [na:1.8.0_101] 
at java.lang.Thread.run(Thread.java:745) [na:1.8.0_101]

嘗試通過設定的TaskExecutor只使用一個線程中運行它的順序。

public TaskExecutor taskExecutor() { 
    SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor(); 
    taskExecutor.setConcurrencyLimit(1); 
    return taskExecutor; 
}

來源

2016-08-01 10:43:21

嗨，設置concurrencyLimit爲1仍然給我同樣的錯誤。但是感謝您指出堆棧跟蹤顯示這與線程有關。如果有人有任何想法，仍然在尋找解決這個問題的辦法。謝謝 – Richie

我一直在玩這個代碼的bean範圍，我認爲它已經讓我在某個地方。現在我有一個關於如何限制我的bean的問題。我想我會關閉這個問題，並開始一個新的範圍問題。感謝您的幫助 – Richie

很高興它爲您提供幫助。關於bean範圍，它與屬性的綁定有關（如在什麼時候），請參閱http://docs.spring.io/spring-batch/reference/html/configureStep.html#step -範圍 –

在第一次閱讀時，讀者需要很長時間才能做好準備。有時rowMapper可能會超前，它認爲讀者沒有打開，當它仍然處於打開過程中時。爲了解決這個問題，只要在你第一次使用rowMapper的時候就睡覺。將下面的塊添加到你的setRowMapper中，看看錯誤是否消失。

try { 
    if(firstTime) { 
     Thread.sleep(20000); 
     firstTime = false; 
    } 
} catch (InterruptedException e) { 
    e.printStackTrace(); 
}

來源

2016-09-09 01:06:23 Yilei

JdbcCursorItemReader的分區程序 - 讀取程序在讀取之前必須打開

回答

相關問題