spring-batch - Spring Batch 3.0 中对 Job Scope bean 的多线程访问

Question

在 Spring Batch 3.0 中，我尝试在分区和多线程步骤(使用 task:executor bean 配置)中对 bean 使用新的 Job Scope 功能，并且在这两种情况下我都遇到了异常

  Caused by: java.lang.IllegalStateException: No context holder available for job scope
        at org.springframework.batch.core.scope.JobScope.getContext(JobScope.java:153)
        at org.springframework.batch.core.scope.JobScope.get(JobScope.java:92)
        at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:338)

但如果我让 bean 步进范围它工作正常。

我注意到 JobSynchronizationManager 上的评论说

N.B. it is the responsibility of every {@link Job} implementation to ensure that a {@link JobContext} is available on every thread that might be involved in a job execution, including worker threads from a pool.

所以我想知道我是否需要做一些事情来设置它，或者它是否是作业范围实现中的一个错误，它没有正确设置工作线程？

StepSynchronizationManager 有类似的评论 - 但在这种情况下，显然在步骤中正确设置了线程。

重现问题的示例代码:

测试项阅读器

package test;

import java.util.ArrayList;
import java.util.List;

import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.NonTransientResourceException;
import org.springframework.batch.item.ParseException;
import org.springframework.batch.item.UnexpectedInputException;
import org.springframework.beans.factory.InitializingBean;

public class TestItemReader implements ItemReader<Integer>, InitializingBean {

    private List<Integer> items;

    @Override
    public synchronized Integer read() throws Exception, UnexpectedInputException,
            ParseException, NonTransientResourceException {

        if (items.size() > 0) {
            return items.remove(0);
        }

        return null;
    }

    @Override
    public void afterPropertiesSet() throws Exception {

        System.out.println("Initialising reader");

        items = new ArrayList<Integer>();   
        for (int i=0;i<100;i++) items.add(i);       
    }
}

测试项编写器

package test;

import java.util.List;

import org.springframework.batch.item.ItemWriter;

public class TestItemWriter implements ItemWriter<Integer> {

    @Override
    public void write(List<? extends Integer> items) throws Exception {

        for (int i : items) {
            System.out.println(Thread.currentThread().getName() + " Writing " + i);
        }       
    }
}

测试作业上下文.xml

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
    xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
    xmlns:batch="http://www.springframework.org/schema/batch"
    xmlns:jdbc="http://www.springframework.org/schema/jdbc"
    xmlns:util="http://www.springframework.org/schema/util"
    xmlns:task="http://www.springframework.org/schema/task"
    xsi:schemaLocation="http://www.springframework.org/schema/batch http://www.springframework.org/schema/batch/spring-batch.xsd
                        http://www.springframework.org/schema/jdbc http://www.springframework.org/schema/jdbc/spring-jdbc.xsd
                        http://www.springframework.org/schema/util http://www.springframework.org/schema/util/spring-util.xsd
                        http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans.xsd
                         http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task.xsd">

    <job id="job" restartable="true" xmlns="http://www.springframework.org/schema/batch">
        <step id="index">   
            <tasklet task-executor="executor">
                <chunk reader="itemReader" writer="itemWriter" commit-interval="5"/>
            </tasklet>
        </step>
    </job>

    <bean id="itemReader" class="test.TestItemReader" scope="job"/>

    <bean id="itemWriter" class="test.TestItemWriter"/>

    <bean id="transactionManager" class="org.springframework.jdbc.datasource.DataSourceTransactionManager">
        <property name="dataSource" ref="dataSource"/>
    </bean>

    <bean class="org.springframework.batch.test.JobLauncherTestUtils">
        <property name="job" ref="job"/>
    </bean>

    <bean id="jobLauncher" class="org.springframework.batch.core.launch.support.SimpleJobLauncher">
        <property name="jobRepository" ref="jobRepository" />
    </bean>

    <batch:job-repository id="jobRepository"/>

    <jdbc:embedded-database id="dataSource" type="HSQL">
        <jdbc:script location="classpath:/org/springframework/batch/core/schema-hsqldb.sql"/>
    </jdbc:embedded-database>

    <task:executor id="executor" queue-capacity="0" pool-size="5"/>

</beans>

求职测试

package test;

import java.util.Collection;

import org.apache.log4j.BasicConfigurator;
import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.junit.BeforeClass;
import org.junit.Test;
import org.springframework.batch.core.JobExecution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.test.JobLauncherTestUtils;
import org.springframework.beans.factory.annotation.Autowired;

import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
import org.springframework.test.context.ContextConfiguration;

@ContextConfiguration(locations={"test-job-context.xml"})
public class JobTest extends AbstractJUnit4SpringContextTests {

    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;

    @BeforeClass
    public static void beforeClassSetup() {    

        BasicConfigurator.configure();
        Logger.getRootLogger().setLevel(Level.WARN);
        Logger.getLogger("org.springframework.batch.core.scope.JobScope").setLevel(Level.DEBUG);
        Logger.getLogger("org.springframework.batch.core.scope.StepScope").setLevel(Level.DEBUG);
    }

    @Test
    public void testJobLaunch() throws Exception {

        JobExecution execution = jobLauncherTestUtils.launchJob();

        System.out.println("After execution "  + execution);

        Collection<StepExecution> stepExecutions = execution.getStepExecutions();
        for (StepExecution stepExecution : stepExecutions) {
            System.out.println("StepExecution " + stepExecution);
        }
    }   
}

运行上述 JUnit 测试将重现该问题。如果您将阅读器上的示波器更改为步进，或移除示波器，测试将正常完成。

Answer 1

这是因为当前执行(在本例中为 JobExecution)存储在 ThreadLocal 中。 (见 org.springframework.batch.core.scope.context.SynchronizationManagerSupport)。话虽如此，为此添加多线程支持似乎并不合理。随意为它创建一个 Jira 问题(如果你愿意，也可以创建一个拉取请求)。