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

Question

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

例如，我有一个32行的平面文件，1行是页眉记录，然后我将其插入页眉表，然后有30条常规记录和1条页脚记录（按顺序），然后在常规记录的记录29中，它失败，出现数据库转换异常（为测试创建的错误），然后它完成，作业状态为失败，这就可以了，但当我查看数据库时，我发现1条记录为页眉，29条记录为常规数据（除了一条有错误的记录外），没有页脚记录，但我希望事务回滚1条记录为页眉，其余29条记录，但在异常回滚后，它们仍在数据库中。

我不知道我是不是错了，spring批处理中的事务不是这样工作的，或者这是我的配置中的错误还是什么。

以下是ClassifierCompositeItemWriter和一个item writer的代码，其他类似于此：

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在评论中所写，它是块大小，我以这种方式添加了它

@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();        
}