Spring Batch 使用 Spring Boot - 从不重试写入错误

Question

我使用 Spring Boot 创建了一个批处理。这里是批处理的主要配置类：

@Configuration
@EntityScan({"my.domain"})
@EnableJpaRepositories({"my.domain"})
@EnableBatchProcessing
public class BatchConfiguration {

    /** Motif d'identification des fichiers d'entrée */
    @Value("${batch.input-file-pattern}")
    private String inputFilePattern;

    @Bean
    public BatchConfigurer configurer( EntityManagerFactory entityManagerFactory ){ 
        return new MapForcedBatchConfigurer( entityManagerFactory ); 
    }

    @Bean
    public Job myJob( JobBuilderFactory jobs, Step step1 ){
        return jobs.get("myJob")
                .incrementer( new RunIdIncrementer() )
                .flow( step1 )
                .end()
                .build();
    }

    @Bean
    public Step step1( StepBuilderFactory stepBuilderFactory, 
            StepExecutionListener stepExecutionListener,
            ItemReader<Input> myReader,
            ItemProcessor<Input, Dto> myProcessor, 
            ItemWriter<Dto> myWriter ){
        return stepBuilderFactory.get("myStep")
                .listener( stepExecutionListener )
                .<Input, Dto> chunk(1)
                .reader( myReader )
                .processor( myProcessor )
                .writer( myWriter )
                .faultTolerant().skipPolicy( new MySkipPolicy() ).retryLimit( 0 )
                .build();
    }

    @Bean
    public StepExecutionListener stepListener() {
        return new MyStepExecutionListener();
    }

    @Bean
    public ItemReader<Input> myReader() throws IOException {
        return new MyItemReader( inputFilePattern );
    }

    @Bean
    public ItemProcessor<Input, Dto> myProcessor(){
        return new MyItemProcessor();
    }

    @Bean
    public ItemWriter<Dto> myWriter(){
        return new MyItemWriter();
    }

}

当处理一个项目发生错误时，记录器会写一条消息，然后批处理处理下一个元素。这正是我想要的。

但是当写入一个项目发生错误时，批处理总是重试一次操作！这样我每个写入错误都有 2 个错误日志。

如何将批处理配置为从不重试错误，无论错误发生在步骤的哪个部分？

This article 解释如下：

[当我们在写入过程中出现跳过时]，框架必须找出导致失败的项目。对于缓存的读取项目列表中的每个项目，它都会启动一个自己的事务。项目由 ItemProcessor 处理，然后由 ItemWriter 写入。如果没有错误，则提交包含一项的 mini-chunk，并继续迭代下一项。我们预计至少有一个可跳过的异常，当这种情况发生时，事务将回滚，并且该项目被标记为已跳过的项目。一旦我们的迭代完成，我们就会继续正常的块处理。

这可能是对重新执行写操作的解释。我希望我能以某种方式绕过这种行为......

Answer 1

我终于想出了一个非常简单的解决方法。在标记为@OnSkipInWrite 的方法中出现错误时，我移动了日志和操作。这样这段代码只执行一次，当 Spring Batch 最终跳过导致错误的元素时。

public class MyItemWriter implements ItemWriter<Dto> {

    @Override
    public void write(List<? extends Dto> items) throws Exception {
        // The writing treatment which may throw a skippable exception...
    }

    @OnSkipInWrite
    public void onSkipInWrite( Dto skippedItem, Exception exception ){
        // Logs, counters, etc.
    }

}

这并不完美，因为抛出异常的处理仍然执行了两次。但我不再有两倍的日志，而且我的自定义错误计数是正确的。

我仍然对抑制这种“通过小块重试一次”行为的方法感兴趣，即使我怀疑它是否可能。

Answer 2

如果不继承 FaultTolerantChunkProcessor，我认为您无能为力。在步骤生成器上调用readerIsTransactionalQueue()会强制执行您想要的行为，但它会忽略所有类似的异常（不会调用您的跳过策略）。

原因是 Spring Batch 试图确定导致错误的项目，即使你的块只有一个项目，算法也没有考虑到它（事实上，你为什么使用大小为 1 的块？ ...)

您可以做的是自己在编写器中捕获异常。您甚至可以为此单独编写一个包装 ItemWriter。