spring-batch - Spring Batch 作业在重新启动时抛出 ItemStreamException；有一个跳过记录的处理器，并且跳过监听器写入平面文件

Question

我需要您的帮助来解决我可能犯的一个非常简单的错误。

我有一个非常基本的 Spring Batch 测试工作。它利用 FlatFileItemReader 从文件中读取并使用 FlatFileItemWriter 进行写入。如果项目的 id 以 a 结尾，则自定义处理器会抛出自定义 SkipObjectException。跳过监听器捕获 SkipObjectException 并利用 FlatFileItemWriter 将消息(字符串)写入文件。

重新启动时，根据失败的时间，我收到 ItemStreamException:当前文件大小小于上次提交的大小。此异常发生在跳过文件上。如果我不写入跳过监听器中的文件，则一切正常。

输入文件内容 001;第一条记录 002；第二条记录 003；第三条记录 003a;第一条跳过记录 004；第四条记录 005；第五条记录 006；第六条记录 007；第七条记录 007a;第二个跳过记录 008；第八条记录 009；第九条记录 010；第十条记录 011；第十一条记录 012；第十二条记录

测试成功 工作进展顺利。它将 12 条记录(除了 003a 和 007a 之外的所有记录)写入输出文件，并将两条记录(003a 和 007a)写入跳过的文件。太棒了。

失败测试 1 – 提交间隔为 5，处理器在项目 004 上失败。正如预期的那样，作业失败，输出文件和跳过文件均为空；这是有道理的，因为异常发生在提交间隔之前。

重新启动故障测试 1 – 与上面相同，但从处理器中删除异常。重新启 Action 业，它会按预期运行，并具有与成功测试相同的输出。

失败测试 2 – 提交间隔为 5；让处理器在项目 004 上失败。结果与之前相同。

重新启动失败测试 2 - 提交间隔为 5；让处理器在项目 008 上失败。作业重新启动，然后在预期的位置失败。输出文件有 4 条记录(001 到 004，003a 除外)，跳跃文件有 1 条记录 003a。因此，输出符合预期。

重新启动失败测试 2 – 提交间隔为 5；从处理器中删除异常。作业无法重新启动，但出现 ItemStreamException 异常:当前文件大小小于上次提交时的大小。此异常发生在跳过文件上。

失败测试 3 – 将提交间隔设置为 10，并使处理器在第 012 项上失败。作业按预期在最后一项上失败。输出文件包含除 003a 和 007a 之外的项目 001 到 008。跳过文件包含003a和007a。一切都好。

重新启动失败测试 3 – 将提交间隔设置为 10 并从处理器中删除异常。当我重新启 Action 业时，我收到 ItemStreamException: 当前文件大小小于上次提交的大小。这发生在跳过输出文件上。

调试信息 我在调试中执行失败测试3。重新启动后，在跳过文件的 FlatFileItemWriter 中，它认为写入了 8 条记录，而实际上应该只有 2 条记录。

我尝试实现 SkipListener、扩展 SkipListenerSupport、使用注释、实现 ItemStream，所有这些都具有相同的结果。是的，我可以使用记录器(例如 log4j)来编写它，但我试图找出我在实现 FlatFileItemWriter 时做错了什么。

就像我之前所说的，这几乎是最简单的，但下面是代码的重要部分 作业.xml

    <batch:step id="step1">
        <batch:tasklet allow-start-if-complete="false">
            <batch:chunk reader="fileReader" processor="fileProcessor"
                writer="fileWriter" commit-interval="5" skip-limit="10">
                <batch:skippable-exception-classes>
                    <batch:include
                        class="com.myexception.SkipObjectException" />
                </batch:skippable-exception-classes>
                <batch:streams>
                    <batch:stream ref="skipListenerWriter" />
                </batch:streams>
            </batch:chunk>
            <batch:listeners>
                <batch:listener ref="customSkipListener" />
            </batch:listeners>
        </batch:tasklet>
    </batch:step>
</batch:job>

<bean id="fileReader" class="org.springframework.batch.item.file.FlatFileItemReader">
    <property name="resource" value="file:TestFiles/input/input.txt" />
    <property name="lineMapper">
        <bean class="org.springframework.batch.item.file.mapping.DefaultLineMapper">
            <property name="lineTokenizer">
                <bean
                    class="org.springframework.batch.item.file.transform.DelimitedLineTokenizer">
                    <property name="delimiter" value=";" />
                </bean>
            </property>
            <property name="fieldSetMapper" ref="fileMapper" />
        </bean>
    </property>
</bean>

<bean id="fileWriter" class="org.springframework.batch.item.file.FlatFileItemWriter">
    <property name="resource" value="file:TestFiles/output/output.txt" />
    <property name="shouldDeleteIfExists" value="true" />
    <property name="lineAggregator">
        <bean
            class="org.springframework.batch.item.file.transform.PassThroughLineAggregator" />
    </property>
</bean>

<bean id="skipListenerWriter" class="org.springframework.batch.item.file.FlatFileItemWriter">
    <property name="resource" value="file:TestFiles/output/skipListener.txt" />
    <property name="shouldDeleteIfExists" value="true" />
    <property name="lineAggregator">
        <bean
            class="org.springframework.batch.item.file.transform.PassThroughLineAggregator" />
    </property>
</bean>

处理器 - 如果正在处理的项目有一个以 a 结尾的 id，则除了抛出 SkipObjectException 之外什么也不做。

自定义跳过监听器

@Autowired
private FlatFileItemWriter<String> skipListenerWriter;

@OnSkipInProcess
public void processLog(FileObject theItem, Throwable theException) {
    String myMessage = "Process skipped item: " + theException.getMessage()
            + " item: " + theItem.getNumber();

    List<String> theItems = new ArrayList<String>();
    theItems.add(myMessage);
    logger.fatal(myMessage);
    try {
        this.skipListenerWriter.write(theItems);
    } catch (Exception e) {
        // TODO wouldn't actually do this...
        throw new RuntimeException(e);
    }
}

Answer 1

我的解决方案，无论正确与否，都是使用 FlatFileItemWriter 扩展的 ItemStreamSupport 抽象类中的 name 属性来唯一命名编写器。

Spring Batch 将写入文件的数量存储在执行上下文中的 name.writing 键下。 FlatFileItemWriter 的默认名称是 FlatFileItemWriter。由于我有两个 FlatFileItemWriter，因此只有一项写入执行上下文 (FlatFileItemWriter.writing)。由于一个文件的写入量比重新启动时发生的其他损坏要多。独特地命名作家解决了这个问题。如果有更合适的方法来处理这个问题，我将不胜感激。