将 Partitioner ExecutionContext 传递给阅读器的问题答案

【问题标题】：Issue in passing Partitioner ExecutionContext to reader将 Partitioner ExecutionContext 传递给阅读器的问题
【发布时间】：2017-01-13 12:52:23
【问题描述】：

我正在尝试实现一个简单的事情 - 将基于 xml 的分区器作业转换为注释。

卡在这部分，

<bean id="pagingItemReader" class="org.springframework.batch.item.database.JdbcPagingItemReader"
        scope="step">


        <property name="parameterValues">
            <map>
                <entry key="minId" value="#{stepExecutionContext[minId]}" />
                <entry key="maxId" value="#{stepExecutionContext[maxId]}" />
            </map>
        </property>

    </bean>

对应的注解代码如，

@Bean
    public Step masterStep(){
        return stepBuilderFactory.get("masterStep")
                .partitioner(slaveStep(syncReader(null,null),writer(),processor()))
                .partitioner("slaveStep", rangePartitioner)
                .gridSize(10)
                .taskExecutor(simpleAsyntaskExecutor)
                .build();
    }

    @Bean(name="slaveStep")
    public Step slaveStep(
            ItemReader<RemittanceVO> syncReader, ItemWriter<RemittanceClaimVO> writer,
            ItemProcessor<RemittanceVO, RemittanceClaimVO> processor) {

        return stepBuilderFactory.get("slaveStep")
                .<RemittanceVO, RemittanceClaimVO> chunk(Constants.SPRING_BATCH_CHUNK_SIZE)
                .reader(syncReader)
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public Partitioner rangePartitioner(){
        RangePartitioner rangePartitioner = new RangePartitioner();
        return rangePartitioner;
    }

@Bean
    @StepScope
    public ItemReader<RemittanceVO> syncReader(@Value("#{stepExecutionContext[minId]}") Long minId,@Value("#{stepExecutionContext[maxId]}") Long maxId) {
        SynchronizedItemStreamReader<RemittanceVO> syncReader = new SynchronizedItemStreamReader<RemittanceVO>();
        syncReader.setDelegate(reader(minId,maxId));
        return syncReader;
    }

    @Bean
    @StepScope
    public ItemStreamReader<RemittanceVO> reader(Long minId,Long maxId) {
        JdbcPagingItemReader<RemittanceVO> reader = new JdbcPagingItemReader<RemittanceVO>();
        reader.setDataSource(dataSource);
        reader.setRowMapper(new RemittanceRowMapper());
        reader.setQueryProvider(queryProvider);
        reader.setPageSize(Constants.SPRING_BATCH_READER_PAGE_SIZE);
        Map<String,Object> parameterValues = new HashMap<String,Object>();
        if(minId !=null) parameterValues.put("minId", minId);
        if(maxId !=null) parameterValues.put("maxId", maxId);
        reader.setParameterValues(parameterValues);
        return reader;
    }

    @Bean
    @StepScope
    public ItemProcessor<RemittanceVO, RemittanceClaimVO> processor() {
        return new MatchClaimProcessor();
    }

    @Bean
    @StepScope
    public ItemWriter<RemittanceClaimVO> writer() {
        return new MatchedClaimWriter();
    }

我收到此代码错误，似乎 Long 依赖不满足，

35307 [SimpleAsyncTaskExecutor-6] ERROR - 2016-09-06 12:23:48.692; org.springframework.batch.core.step.AbstractStep; Encountered an error executing step slaveStep in job runPartitionerJob
org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'scopedTarget.reader' defined in class path resource [com/hmsy/bp/config/AppConfiguration.class]: Unsatisfied dependency expressed through method 'reader' parameter 0: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:749)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:467)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1123)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1018)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:510)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:482)
    at org.springframework.beans.factory.support.AbstractBeanFactory$2.getObject(AbstractBeanFactory.java:345)
    at org.springframework.batch.core.scope.StepScope.get(StepScope.java:113)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:340)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:197)
    at org.springframework.aop.target.SimpleBeanTargetSource.getTarget(SimpleBeanTargetSource.java:35)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:192)
    at com.sun.proxy.$Proxy62.read(Unknown Source)
    at org.springframework.batch.item.support.SynchronizedItemStreamReader.read(SynchronizedItemStreamReader.java:55)
    at sun.reflect.GeneratedMethodAccessor52.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:133)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:121)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
    at com.sun.proxy.$Proxy59.read(Unknown Source)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.doRead(SimpleChunkProvider.java:91)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.read(SimpleChunkProvider.java:157)
    at org.springframework.batch.core.step.item.SimpleChunkProvider$1.doInIteration(SimpleChunkProvider.java:116)
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.provide(SimpleChunkProvider.java:110)
    at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:69)
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:406)
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:330)
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133)
    at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271)
    at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:81)
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
    at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200)
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:139)
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:136)
    at java.util.concurrent.FutureTask.run(Unknown Source)
    at org.springframework.core.task.SimpleAsyncTaskExecutor$ConcurrencyThrottlingRunnable.run(SimpleAsyncTaskExecutor.java:268)
    at java.lang.Thread.run(Unknown Source)
Caused by: org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.raiseNoMatchingBeanFound(DefaultListableBeanFactory.java:1406)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1057)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1019)
    at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:835)
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:741)
    ... 47 more
35307 [SimpleAsyncTaskExecutor-11] ERROR - 2016-09-06 12:23:48.692; org.springframework.batch.core.step.AbstractStep; Encountered an error executing step slaveStep in job runPartitionerJob
org.springframework.beans.factory.UnsatisfiedDependencyException: Error creating bean with name 'scopedTarget.reader' defined in class path resource [com/hmsy/bp/config/AppConfiguration.class]: Unsatisfied dependency expressed through method 'reader' parameter 0: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}; nested exception is org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:749)
    at org.springframework.beans.factory.support.ConstructorResolver.instantiateUsingFactoryMethod(ConstructorResolver.java:467)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.instantiateUsingFactoryMethod(AbstractAutowireCapableBeanFactory.java:1123)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBeanInstance(AbstractAutowireCapableBeanFactory.java:1018)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:510)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:482)
    at org.springframework.beans.factory.support.AbstractBeanFactory$2.getObject(AbstractBeanFactory.java:345)
    at org.springframework.batch.core.scope.StepScope.get(StepScope.java:113)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:340)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:197)
    at org.springframework.aop.target.SimpleBeanTargetSource.getTarget(SimpleBeanTargetSource.java:35)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:192)
    at com.sun.proxy.$Proxy62.read(Unknown Source)
    at org.springframework.batch.item.support.SynchronizedItemStreamReader.read(SynchronizedItemStreamReader.java:55)
    at sun.reflect.GeneratedMethodAccessor52.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(Unknown Source)
    at java.lang.reflect.Method.invoke(Unknown Source)
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:333)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:190)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:157)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:133)
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:121)
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:179)
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:213)
    at com.sun.proxy.$Proxy59.read(Unknown Source)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.doRead(SimpleChunkProvider.java:91)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.read(SimpleChunkProvider.java:157)
    at org.springframework.batch.core.step.item.SimpleChunkProvider$1.doInIteration(SimpleChunkProvider.java:116)
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
    at org.springframework.batch.core.step.item.SimpleChunkProvider.provide(SimpleChunkProvider.java:110)
    at org.springframework.batch.core.step.item.ChunkOrientedTasklet.execute(ChunkOrientedTasklet.java:69)
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:406)
    at org.springframework.batch.core.step.tasklet.TaskletStep$ChunkTransactionCallback.doInTransaction(TaskletStep.java:330)
    at org.springframework.transaction.support.TransactionTemplate.execute(TransactionTemplate.java:133)
    at org.springframework.batch.core.step.tasklet.TaskletStep$2.doInChunkContext(TaskletStep.java:271)
    at org.springframework.batch.core.scope.context.StepContextRepeatCallback.doInIteration(StepContextRepeatCallback.java:81)
    at org.springframework.batch.repeat.support.RepeatTemplate.getNextResult(RepeatTemplate.java:374)
    at org.springframework.batch.repeat.support.RepeatTemplate.executeInternal(RepeatTemplate.java:215)
    at org.springframework.batch.repeat.support.RepeatTemplate.iterate(RepeatTemplate.java:144)
    at org.springframework.batch.core.step.tasklet.TaskletStep.doExecute(TaskletStep.java:257)
    at org.springframework.batch.core.step.AbstractStep.execute(AbstractStep.java:200)
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:139)
    at org.springframework.batch.core.partition.support.TaskExecutorPartitionHandler$1.call(TaskExecutorPartitionHandler.java:136)
    at java.util.concurrent.FutureTask.run(Unknown Source)
    at org.springframework.core.task.SimpleAsyncTaskExecutor$ConcurrencyThrottlingRunnable.run(SimpleAsyncTaskExecutor.java:268)
    at java.lang.Thread.run(Unknown Source)
Caused by: org.springframework.beans.factory.NoSuchBeanDefinitionException: No qualifying bean of type [java.lang.Long] found for dependency [java.lang.Long]: expected at least 1 bean which qualifies as autowire candidate for this dependency. Dependency annotations: {}
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.raiseNoMatchingBeanFound(DefaultListableBeanFactory.java:1406)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.doResolveDependency(DefaultListableBeanFactory.java:1057)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.resolveDependency(DefaultListableBeanFactory.java:1019)
    at org.springframework.beans.factory.support.ConstructorResolver.resolveAutowiredArgument(ConstructorResolver.java:835)
    at org.springframework.beans.factory.support.ConstructorResolver.createArgumentArray(ConstructorResolver.java:741)
    ... 47 more

正如在this question 中所做的那样，我从masterStep 传递null。 masterStep bean 创建成功，尝试执行masterStep 时出错。

XML 代码可以正常工作。我只显示了部分 xml。

【问题讨论】：

标签： spring-boot spring-batch

【解决方案1】：

我能够通过将非同步阅读器代码移动到同步装饰器代码来解决问题。不知何故，从同步阅读器向非同步阅读器调用和传递参数不起作用。

@Bean
    @StepScope
    public ItemReader<RemittanceVO> syncReader(@Value("#{stepExecutionContext[minId]}") Long minId,@Value("#{stepExecutionContext[maxId]}") Long maxId) throws Exception{
        SynchronizedItemStreamReader<RemittanceVO> syncReader = new SynchronizedItemStreamReader<RemittanceVO>();
        Map<String,Object> parameterValues = new HashMap<String,Object>();
        if(minId !=null && minId.longValue() !=0)  parameterValues.put("minId", minId);
        if(maxId !=null && maxId.longValue() !=0) parameterValues.put("maxId", maxId);
        JdbcPagingItemReader<RemittanceVO> reader = new JdbcPagingItemReader<RemittanceVO>();
        reader.setDataSource(dataSource);
        reader.setRowMapper(new RemittanceRowMapper());
        reader.setQueryProvider(queryProvider);
        reader.setPageSize(Constants.SPRING_BATCH_READER_PAGE_SIZE);
        reader.setParameterValues(parameterValues);
        reader.afterPropertiesSet();
        syncReader.setDelegate(reader);
        return syncReader;
    }

【讨论】：

您确定要同步阅读器吗？这将消除对分区步骤进行多线程处理的能力。
我是批处理 API 的新手，对大部分事情都不太了解。在回答我的another question 之一时，建议使用它，但它不在分区上下文中。
只要您的阅读器在步骤范围内（它必须是为了访问步骤执行上下文），那么您将在每个步骤中拥有一个实例（因此每个步骤都有一个实例线程线）。假设读者没有操纵静态属性，每个线程有一个实例将保证您是“线程安全的”。在这种情况下进行同步会大大减慢您的速度，而没有任何好处。