java - 跳过 ItemProcessor 中的异常会导致二次行为

Question

有没有办法改变 ItemProcessor 的跳过行为，使其表现得像 ItemWriter？在 ItemProcessor 中抛出可跳过的异常会导致重新处理所有已接受的项目，从而导致其复杂性呈二次方的行为。有没有办法改变行为以对所有项目执行回滚并逐个处理元素，就像 ItemWriter 一样？

工作定义:

@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]

在示例中，项目 0-4 在成功写入之前被处理了 6 次(假设 Writer 不会抛出)。

• 5 次因为处理器回滚
• 1 表示处理成功

如果 ItemProcessor 使用与 ItemWriter 相同的跳过策略，那么它们只会被处理 2 次:

• 处理失败1次
• 1 次单独交易

Answer 1

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

@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 处理项目时，我觉得在回滚的情况下更安全，在回滚事务中加载和处理的所有实体都应该被丢弃，最好不要在另一个新事务中重用它们，因为实体将在新交易对我来说似乎让事情变得更复杂。

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