Spring Batch多线程下WriterNotOpenException问题及XML配置解决咨询

场景说明 #

项目中使用FlatFileItemReader结合Classifier与ClassifierCompositeItemWriter，读取制表符分隔文件，经校验、增强后转为CSV格式输出，通过Classifier逻辑生成4个输出文件。运行时抛出WriterNotOpenException错误，需通过XML配置触发itemWriter.open(new ExecutionContext())解决。

现有配置代码 #

<bean id="classifier" class="com.XXXXXXXXX.SomeClassifier">
    <constructor-arg index="0" ref="oneMultiResource"/>
    <constructor-arg index="1" ref="twoMultiResource"/>
    <constructor-arg index="2" ref="threeMultiResource"/>
    <constructor-arg index="3" ref="fourMultiResource"/>
</bean>

<bean id="classifierCompositeItemWriter" class="org.springframework.batch.item.support.ClassifierCompositeItemWriter">
    <property name="classifier" ref="classifier"/>
</bean>

<bean id="onetCvsWriter" class="org.springframework.batch.item.file.FlatFileItemWriter" scope="step" >
    <property name="resource" value="file:${XXXXXXXXXXXXXXXXX}.#{jobParameters['year']}"/>
    <property name="shouldDeleteIfExists" value="true"/>
    <property name="shouldDeleteIfEmpty" value="true" />
    <property name="appendAllowed" value="true" />
    <property name="encoding" value="UTF-8"/>
    <property name="lineAggregator" ref="abLineAggregator"/>
    <property name="headerCallback" ref="headerCallback" />
</bean>

<bean id="oneMultiResource" class="org.springframework.batch.item.file.MultiResourceItemWriter" scope="step">
    <property name="name" value="MultiResourceWriter" />
    <property name="delegate" ref="onetCvsWriter" />
    <property name="resource" value="file:${XXXXXXXXXXXXXXXXX}.#{jobParameters['year']}" />
    <property name="itemCountLimitPerResource" value="500000"  />
    <property name="resourceSuffixCreator" ref="suffixCreator" />
</bean>

错误日志 #

org.springframework.batch.item.WriterNotOpenException: Writer must be open before it can be written to
    at org.springframework.batch.item.file.FlatFileItemWriter.write(FlatFileItemWriter.java:236) ~[spring-batch-infrastructure-2.1.8.RELEASE.jar:?]
    at sun.reflect.GeneratedMethodAccessor16.invoke(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_202]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_202]
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:137) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:124) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:215) ~[spring-aop-5.3.21.jar:5.3.21]
    at com.sun.proxy.$Proxy56.write(Unknown Source) ~[?:?]
    at org.springframework.batch.item.file.MultiResourceItemWriter.write(MultiResourceItemWriter.java:82) ~[spring-batch-infrastructure-2.1.8.RELEASE.jar:?]
    at sun.reflect.GeneratedMethodAccessor16.invoke(Unknown Source) ~[?:?]
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_202]
    at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_202]
    at org.springframework.aop.support.AopUtils.invokeJoinpointUsingReflection(AopUtils.java:344) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.invokeJoinpoint(ReflectiveMethodInvocation.java:198) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:163) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.doProceed(DelegatingIntroductionInterceptor.java:137) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.support.DelegatingIntroductionInterceptor.invoke(DelegatingIntroductionInterceptor.java:124) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:186) ~[spring-aop-5.3.21.jar:5.3.21]
    at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:215) ~[spring-aop-5.3.21.jar:5.3.21]
    at com.sun.proxy.$Proxy54.write(Unknown Source) ~[?:?]

问题原因 #

Classifier直接注入了MultiResourceItemWriter实例，但这些writer的生命周期方法（open/close）未被Spring Batch框架自动触发——ClassifierCompositeItemWriter仅通过classifier获取writer并调用write，不会主动管理其生命周期，导致写入时writer未处于打开状态。

XML配置解决方案 #

方案1：通过MethodInvokingFactoryBean手动调用open方法 #

直接在XML中配置调用每个writer的open方法，绑定到step初始化流程：

<!-- 调用oneMultiResource的open方法 -->
<bean id="openOneMultiResource" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
    <property name="targetObject" ref="oneMultiResource"/>
    <property name="targetMethod" value="open"/>
    <property name="arguments">
        <list>
            <bean class="org.springframework.batch.item.ExecutionContext"/>
        </list>
    </property>
</bean>

<!-- 为另外三个writer创建同样的配置 -->
<bean id="openTwoMultiResource" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
    <property name="targetObject" ref="twoMultiResource"/>
    <property name="targetMethod" value="open"/>
    <property name="arguments">
        <list>
            <bean class="org.springframework.batch.item.ExecutionContext"/>
        </list>
    </property>
</bean>

<bean id="openThreeMultiResource" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
    <property name="targetObject" ref="threeMultiResource"/>
    <property name="targetMethod" value="open"/>
    <property name="arguments">
        <list>
            <bean class="org.springframework.batch.item.ExecutionContext"/>
        </list>
    </property>
</bean>

<bean id="openFourMultiResource" class="org.springframework.beans.factory.config.MethodInvokingFactoryBean">
    <property name="targetObject" ref="fourMultiResource"/>
    <property name="targetMethod" value="open"/>
    <property name="arguments">
        <list>
            <bean class="org.springframework.batch.item.ExecutionContext"/>
        </list>
    </property>
</bean>

<!-- 在step配置中添加依赖，确保初始化时执行open操作 -->
<bean id="yourStep" class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
    <property name="itemReader" ref="yourFlatFileItemReader"/>
    <property name="itemProcessor" ref="yourProcessor"/>
    <property name="itemWriter" ref="classifierCompositeItemWriter"/>
    <property name="dependsOn">
        <list>
            <value>openOneMultiResource</value>
            <value>openTwoMultiResource</value>
            <value>openThreeMultiResource</value>
            <value>openFourMultiResource</value>
        </list>
    </property>
</bean>

方案2：自定义StepExecutionListener管理生命周期（推荐） #

通过监听器在step启动/结束时调用writer的open/close方法，使用step真实的ExecutionContext，避免状态丢失：

1. 编写监听器Java类

public class WriterOpenListener implements StepExecutionListener {

    private List<ItemStream> writers;

    @Override
    public void beforeStep(StepExecution stepExecution) {
        ExecutionContext executionContext = stepExecution.getExecutionContext();
        for (ItemStream writer : writers) {
            writer.open(executionContext);
        }
    }

    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        ExecutionContext executionContext = stepExecution.getExecutionContext();
        for (ItemStream writer : writers) {
            writer.close();
        }
        return stepExecution.getExitStatus();
    }

    // 注入writer列表的setter
    public void setWriters(List<ItemStream> writers) {
        this.writers = writers;
    }
}

2. XML配置监听器与step

<!-- 注册自定义监听器 -->
<bean id="writerOpenListener" class="com.XXXXXXXXX.WriterOpenListener">
    <property name="writers">
        <list>
            <ref bean="oneMultiResource"/>
            <ref bean="twoMultiResource"/>
            <ref bean="threeMultiResource"/>
            <ref bean="fourMultiResource"/>
        </list>
    </property>
</bean>

<!-- Step配置中注入监听器 -->
<bean id="yourStep" class="org.springframework.batch.core.step.factory.SimpleStepFactoryBean">
    <property name="itemReader" ref="yourFlatFileItemReader"/>
    <property name="itemProcessor" ref="yourProcessor"/>
    <property name="itemWriter" ref="classifierCompositeItemWriter"/>
    <property name="stepExecutionListeners">
        <list>
            <ref bean="writerOpenListener"/>
        </list>
    </property>
</bean>

总结 #

方案2更符合Spring Batch的生命周期管理规范，能正确维护writer的状态；方案1适合快速临时修复。两种方式都能解决WriterNotOpenException问题，确保writer在写入前处于打开状态。

内容的提问来源于stack exchange，提问作者PAA