ClassifierCompositeItemWriter 异常回滚，但数据部分提交到数据库

Question

我正在使用 ClassifierCompositeItemWriter 插入存储在一个固定长度平面文件中的不同类型的寄存器，并将其写入 具有多个 JdbcBatchItemWriters 的 postgres 数据库，每个 JdbcBatchItemWriters 在一个不同的表中，所有这一切都在一个步骤中然后坚持一个弹簧批处理作业，它工作正常，但是在激活事务时它们不会在发生异常时回滚。

例如我有一个 32 行的平面文件，其中 1 行是表头记录，然后我将其插入表头表中，然后有 30 条常规记录和 1 条表尾记录（按顺序），然​​后在第 29 条记录中常规的，它因数据库转换异常（为测试而创建的错误）而失败，然后以作业的失败状态结束，这没关系，但是当我查看数据库时，我发现标题的 1 条记录，常规数据的 29 条记录（除了有错误的一条）并且没有页脚记录，但我希望事务回滚该 1 条记录的页眉和其他 29 条记录，但异常回滚后它们仍在数据库中。

我不知道我是否错了，春季批处理中的事务不能以这种方式工作，或者是我的配置错误还是什么。

这是 ClassifierCompositeItemWriter 的代码，其他一个项目编写器与此类似：

public ClassifierCompositeItemWriter<DTOBase> altasOffWriterClassifier(DataSource dataSource) {

    BackToBackPatternClassifier classifier = new BackToBackPatternClassifier();
    classifier.setRouterDelegate(dtoWriterClassifier);
    classifier.setMatcherMap(new HashMap<String, JdbcBatchItemWriter<? extends DTOBase>>() {
        private static final long serialVersionUID = -1247536568421993759L;
    {
        put(DTOHeader.class.getTypeName(), headerWriter());
        put(DTOData.class.getTypeName(), dataWriter());
        put(DTOFooter.class.getTypeName(), footerWriterFin());
    }});

    ClassifierCompositeItemWriter<DTOBase> writer = new ClassifierCompositeItemWriter<>();
    writer.setClassifier(classifier);

    return writer;
}

@Bean
public JdbcBatchItemWriter<DTOAltaOFF> altaOffWriter() {
    return new JdbcBatchItemWriterBuilder<DTOData>()
         .dataSource(dataSource)
         .sql("insert into tabla(ticket, identificador, fecha_alta_operacion, "
                + " ordenante, numero, moneda, cif, importe_emisor, "
                + " estado, telefono_destino, fecha_caducidad_hal, concepto, cabecera_num_orden_fichero) "
                + " VALUES (:ticket,:identificador,to_timestamp(:fechaAltaOperacion,'DDMMYYYYHH24MISS'), "
                + " :ordenante,:numero,:moneda,:cif,(cast(:importeEmisor as double precision)/100), "
                + " :estado,:telefonoDestino,to_timestamp(:fechaCaducidadHal,'DDMMYYYYHH24MISS'),:concepto,:idCabecera) ")
         .itemSqlParameterSourceProvider(new BeanPropertyItemSqlParameterSourceProvider<>()) 
         .build();     
}

我的配置类：

@Configuration
@EnableBatchProcessing
@Import({ DataSourceConfig.class })
@PropertySource("classpath:batch.properties")
@ComponentScan({ "..."})
public class BatchConfiguration {

@Autowired
public JobBuilderFactory jobBuilderFactory;

@Autowired
public StepBuilderFactory stepBuilderFactory;

@Autowired
public JobRepository jobRepository;

@Autowired
public DataSource dataSource;

@Bean
public JdbcTemplate getJdbcTemplate() {
    return new JdbcTemplate(dataSource);
}

@Bean
public TaskExecutor taskExecutor() {
    return new SimpleAsyncTaskExecutor();
}

数据源

@Configuration
@EnableTransactionManagement
public class DataSourceConfig {

...some @Value...   
    @Bean(name = "dataSource")
    public DriverManagerDataSource dataSource() {
        DriverManagerDataSource dataSource = new DriverManagerDataSource();
        dataSource.setDriverClassName(driverClassName);
        dataSource.setUrl(datasourceUrl);
        dataSource.setUsername(usuario);
        dataSource.setPassword(clave);
        return dataSource;
    }
}

配置：

@Configuration
@EnableBatchProcessing
@Import({ DataSourceConfig.class })
@PropertySource("classpath:batch.properties")
@ComponentScan({ "..."})
public class BatchConfiguration {

    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    @Autowired
    public JobRepository jobRepository;

    @Autowired
    public DataSource dataSource;

    @Bean
    public JdbcTemplate getJdbcTemplate() {
        return new JdbcTemplate(dataSource);
    }

    @Bean
    public TaskExecutor taskExecutor() {
        return new SimpleAsyncTaskExecutor();
    }
}

自定义：

@Component
@EnableTransactionManagement
public class CustomBatchConfigurer extends DefaultBatchConfigurer {

   private final TaskExecutor taskExecutor;

    public CustomBatchConfigurer(TaskExecutor taskExecutor) {
        this.taskExecutor = taskExecutor;
    }

    @Override
    protected JobLauncher createJobLauncher() throws Exception {
        SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
        jobLauncher.setJobRepository(getJobRepository());
        jobLauncher.setTaskExecutor(this.taskExecutor);
        jobLauncher.afterPropertiesSet();
        return jobLauncher;
    }

    @Autowired
    private DataSource dataSource;

    @Override
    public PlatformTransactionManager getTransactionManager() {
        DataSourceTransactionManager tm = new DataSourceTransactionManager();
        tm.setDataSource(dataSource);
        return tm;
    }
}

任何帮助都会很棒。

Answer 1

这是@MahmoudBenHassine 在 cmets 中写的块大小，我以这种方式添加它

@Bean
public Step Step1(@Qualifier("xyzWriterClassifier") ItemWriter<DTOxyz> writer) throws Exception {
    return stepBuilderFactory.get("Step1")
            .<DTOxyz, DTOxyz> chunk(100)
            .reader(dtoXyzItemReader(NULL because WILL_BE_INJECTED))
            .processor(XyzProcessor())
            .writer(writer)
            .build();        
}