Spring Batch中ItemWriter异常后如何实现数据持久化？

这确实是Spring Batch里很常见的一个痛点——默认的Chunk事务机制会在ItemWriter抛出异常时回滚整个Chunk的所有操作，包括你想标记为错误状态的那些数据，所以直接在ItemWriteListener的onWriteError里写持久化逻辑，会跟着原事务一起回滚，导致不生效。不过有几种可行的方案能解决这个问题，咱们来详细说：

方案一：给错误处理逻辑开启独立事务 #

核心思路是让你标记错误状态的逻辑脱离原Chunk的事务，这样即使原事务回滚，这部分操作也能独立提交。具体可以通过声明式事务的REQUIRES_NEW传播属性来实现：

1. 首先写一个专门处理错误状态更新的Service，给方法加上独立事务：

@Service
public class ErrorItemHandler {

    @Autowired
    private YourEntityRepository entityRepository;

    // 开启新事务，不受原Chunk事务回滚影响
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public void markItemsAsFailed(List<YourEntity> failedItems, Exception error) {
        failedItems.forEach(item -> {
            item.setStatus("FAILED");
            item.setErrorMsg(error.getMessage());
            entityRepository.save(item);
        });
    }
}

2. 然后在你的ItemWriteListener里调用这个Service：

@Component
public class CustomItemWriteListener implements ItemWriteListener<YourEntity> {

    @Autowired
    private ErrorItemHandler errorItemHandler;

    @Override
    public void onWriteError(Exception exception, List<? extends YourEntity> items) {
        // 调用独立事务的方法，确保标记操作能提交
        errorItemHandler.markItemsAsFailed(new ArrayList<>(items), exception);
    }
}

3. 最后把这个Listener注册到你的Step中：

@Bean
public Step dataProcessingStep(ItemReader<YourEntity> reader,
                               ItemProcessor<YourEntity, YourEntity> processor,
                               ItemWriter<YourEntity> writer,
                               CustomItemWriteListener writeListener) {
    return stepBuilderFactory.get("dataProcessingStep")
            .<YourEntity, YourEntity>chunk(50) // 这里是你的Chunk大小
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .listener(writeListener)
            .build();
}

这个方案适合当整个Chunk因为某个异常失败时，需要把该Chunk内所有项都标记为错误的场景。

方案二：利用Skip机制单独处理错误项 #

如果你的需求是跳过错误项，继续处理Chunk内其他正常项，同时单独持久化错误项的状态，那可以用Spring Batch的Skip机制配合SkipListener：

1. 先定义一个SkipPolicy，指定哪些异常可以跳过：

@Bean
public SkipPolicy customSkipPolicy() {
    return (throwable, skipCount) -> {
        // 根据实际业务，判断哪些异常需要跳过（比如自定义业务异常）
        return throwable instanceof YourBusinessException;
    };
}

2. 编写SkipListener来处理写入时跳过的项：

@Component
public class CustomSkipListener implements SkipListener<YourEntity, YourEntity> {

    @Autowired
    private ErrorItemHandler errorItemHandler;

    @Override
    public void onSkipInWrite(YourEntity item, Throwable t) {
        // 单独标记跳过的错误项，同样用独立事务
        errorItemHandler.markItemsAsFailed(Collections.singletonList(item), (Exception) t);
    }
}

3. 配置Step时开启容错并注册SkipPolicy和SkipListener：

@Bean
public Step dataProcessingStep(ItemReader<YourEntity> reader,
                               ItemProcessor<YourEntity, YourEntity> processor,
                               ItemWriter<YourEntity> writer,
                               CustomSkipListener skipListener) {
    return stepBuilderFactory.get("dataProcessingStep")
            .<YourEntity, YourEntity>chunk(50)
            .reader(reader)
            .processor(processor)
            .writer(writer)
            .faultTolerant()
            .skipPolicy(customSkipPolicy())
            .listener(skipListener)
            .build();
}

这个方案的优势是不会因为单个项的错误导致整个Chunk失败，正常项可以正常提交，错误项单独标记。

为什么你原来的方式不生效？ #

要注意的是，默认情况下ItemWriteListener的onWriteError方法是在当前Chunk的事务上下文里执行的，当ItemWriter抛出异常，整个Chunk的事务会触发回滚，所以你在这个方法里做的任何持久化操作都会被一起回滚。这就是为什么必须用独立事务来脱离原事务的原因。

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