在ItemProcessor中跳过异常导致二次行为

Question

@Bean
public Job job() {
    return jobBuilderFactory.get("job").start(step(null)).build();
}

@Bean
public Step step(ReaderProcessorWriter readerProcessorWriter) {
    return stepBuilderFactory.get("step")
            .<Integer, Integer>chunk(20).faultTolerant()
            .skip(RuntimeException.class).skipLimit(10)
            .reader(readerProcessorWriter)
            .processor(readerProcessorWriter)
            .writer(readerProcessorWriter)
            .build();
}

@Component
public static class ReaderProcessorWriter extends ListItemReader<Integer> implements ItemProcessor<Integer, Integer>, ItemWriter<Integer> {
    private int run;

    public ReaderProcessorWriter() {
        super(List.of(0, 1, 2, 3, 4, 5, 6, 7,8 ,9));
    }

    @Override
    public Integer process(Integer integer) {
        // Probably some long computation involving lots of DB Reading, lasting minutes at worst
        if (integer >= 5) {
            System.out.println("FAIL: " + integer);
            throw new RuntimeException("Hue hue");
        }
        System.out.println("OK: " + integer);
        return integer;
    }

    @Override
    public void write(List<? extends Integer> list) {
        if (run++ == 0) {
            throw new RuntimeException("Writer");
        }
        System.out.println(list);
    }
}

OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
FAIL: 5
OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
FAIL: 6
OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
FAIL: 7
OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
FAIL: 8
OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
FAIL: 9
OK: 0
OK: 1
OK: 2
OK: 3
OK: 4
OK: 0
[0]
OK: 1
[1]
OK: 2
[2]
OK: 3
[3]
OK: 4
[4]

null

null

Answer 1

您可以尝试将容错处理器配置为非事务性的。然后，它将缓存成功处理的项的处理结果，以便在回滚的情况下和在再次处理块的重试期间，它只从缓存中获得处理结果，而不是再次处理成功的项。

@Bean
public Step step(ReaderProcessorWriter readerProcessorWriter) {
    return stepBuilderFactory.get("step")
            .<Integer, Integer>chunk(20).faultTolerant()
            .skip(RuntimeException.class).skipLimit(10)
            .reader(readerProcessorWriter)
            .processor(readerProcessorWriter)
            .processorNonTransactional()
            .writer(readerProcessorWriter)
            .build();
}

您必须看看它是否适合您的用例。当使用JPA处理这些项时，我觉得更安全的是，在回滚的情况下，应该丢弃在回滚事务中加载和处理的所有实体，最好不要在另一个新事务中重用它们，因为实体将在新事务中分离，这对我来说似乎使事情变得更加复杂。

如果重新处理一个项目需要相当长的时间，我将尝试看看处理一个项目中的瓶颈操作是否可以微调，比如使用缓存来缓存那些扩展操作的结果。