spring批处理、JpaRepository和回滚

Question

我有一个spring批处理应用程序（spring boot 2.3.5版），它在spring批处理时使用一个JpaRepository将一些自定义日志消息插入数据库。这与开箱即用的spring批处理表是分开的。似乎当我从ItemProcessorAdapter抛出异常时，它会被ItemProcessListener onProcessError（）方法捕获。在这个方法中，我执行一个JpaRepository save（）和flush（）。不会记录任何错误，但是一旦我离开这个方法，JpaRepository就会回滚。

1. 这是正常行为吗？我怎样才能避开它？
2. 在使用JpaRepository时，是否有方法设置@transactional(noRollbackFor={xxxException.class}）？我试了一下，似乎没有效果。

下面是示例代码段。

java prettyprint-override">
@Configuration
public class BatchJobConfiguration {

   //Omitted for clarity....

    @Bean
    @StepScope
    public CompositeItemProcessor<Decision,Decision> itemProcessor() {
        CompositeItemProcessor<Decision,Decision> itemProcessor = new CompositeItemProcessor<>();
        itemProcessor.setDelegates(Arrays.asList(
                decisionValidatingItemProcessor(),
                myItemProcessor(null)
        ));
        return itemProcessor;
    } // end itemProcessor()
    
     @Bean
    public BeanValidatingItemProcessor<Decision> decisionValidatingItemProcessor() {
        BeanValidatingItemProcessor<Decision> beanValidatingItemProcessor = new BeanValidatingItemProcessor<>();
        beanValidatingItemProcessor.setFilter(true);
        return beanValidatingItemProcessor;
    } // end decisionValidatingItemProcessor()

    @Bean
    public ItemProcessorAdapter<Decision,Decision> myItemProcessor(DecisionProcessingService service) {
        ItemProcessorAdapter<Decision,Decision> adapter = new ItemProcessorAdapter<>();
        adapter.setTargetObject(service);
        adapter.setTargetMethod("processDecision");
        return adapter;
    }
    
    @Bean
    @StepScope
    public DecisionItemProcessListener decisionItemProcessListener() {
        return new DecisionItemProcessListener(mpJpaRepository);
    }


}

@Service
public class DecisionProcessingService {

    public Decision processDecision(Decision decision) throws BatchException {
        ....
        throw new BatchException("An error occurred");
    
    }

}



public class DecisionItemProcessListener implements ItemProcessListener<Decision,Decision> {
    private MyJpaRepository mpJpaRepository;

    public DecisionItemProcessListener(MyJpaRepository mpJpaRepository) {
        this.mpJpaRepository = mpJpaRepository;
    }
    
    ....
    
    @Override
    public void onProcessError(Decision decision, Exception e) {
        MyEntityObject obj = MyEntityObject.builder()
                .msg(e.getMessage())
                .build();
        mpJpaRepository.save(obj);
        mpJpaRepository.flush();
        
        // after this, the insert above is rolled back.
    } // end onProcessError()
}

Answer 1

您在这里使用的回调ItemProcessListener#OnProcessError是在一个事务（由spring批处理驱动）中调用的，该事务将由于项处理器引发的异常而回滚。

如果要在该方法中保存数据，则需要使用新事务（使用requires_new传播）。

编辑：我在这里分享了一个最小的完整示例:https://github.com/benas/spring-batch-lab/tree/master/issues/so64913980。