使用SimpleAsyncTaskExecutor时JdbcPagingItemReader丢失记录问题排查

问题场景 #

单线程模式下可正常读取所有数据库记录并写入XML文件，但启用多线程（SimpleAsyncTaskExecutor，并发数5）后，700条记录中会丢失4-5条，需保留多线程方案以处理大数据量。

现有配置整理 #

任务执行器配置 #

@Bean
public TaskExecutor tvlTaskExecutor() {
    SimpleAsyncTaskExecutor asyncTaskExecutor = new SimpleAsyncTaskExecutor();
    asyncTaskExecutor.setConcurrencyLimit(5);
    return asyncTaskExecutor;
}

步骤配置 #

@Bean
public Step tvlFileCreationStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
    return new StepBuilder("tvl-file-creation-step", jobRepository)
            .<String, String>chunk(50, transactionManager)
            .reader(tvlTableDataItemReader())
            .writer(tvlFileWriter)
            .listener(tvlStepExecutionListener)
            .taskExecutor(tvlTaskExecutor())
            .build();
}

任务配置 #

@Bean(name="tvlRunJob")
public Job tvlRunJob( JobRepository jobRepository,  Step tvlFileCreationStep) {
    return new JobBuilder("tvlFileCreationJob", jobRepository)
            .start(tvlFileCreationStep)
            .build();
}

读取器配置 #

public JdbcPagingItemReader<String> jdbcPagingItemReader(DataSource dataSource) throws Exception {
    // reading ITAG keys records using JDBC in a paging fashion
    JdbcPagingItemReader<String> reader = new JdbcPagingItemReader<String>();
    reader.setDataSource(dataSource);
    reader.setPageSize(50);
    reader.setSaveState(true);
    reader.setQueryProvider(createQuery(dataSource));
    reader.setRowMapper(new TVLRowMapper());
    return reader;
}

private PagingQueryProvider createQuery(DataSource dataSource) throws Exception {
    final Map<String, Order> sortKeys = new HashMap<>();
    sortKeys.put("TAG_SERIAL_NUMBER", Order.ASCENDING);

    final SqlPagingQueryProviderFactoryBean queryProvider = new SqlPagingQueryProviderFactoryBean();
    queryProvider.setDataSource(dataSource);
    queryProvider.setSelectClause(" TAG_SERIAL_NUMBER ");
    queryProvider.setFromClause(" ITAGENTITY_10052024 ");
    queryProvider.setWhereClause(" TAG_SERIAL_NUMBER like '09900%' ");
    queryProvider.setSortKeys(sortKeys);
    return queryProvider.getObject();
}

写入器配置 #

使用stax2库的XMLOutputFactory2和XMLStreamWriter2，官方文档标注为线程安全实现（包路径：org.codehaus.stax2.*）。

问题根源 #

JdbcPagingItemReader本身不是线程安全的，它内部维护了当前页码、结果集等状态信息。而Spring容器默认将Bean注册为单例模式，多线程Step中的所有线程会共享同一个Reader实例，导致多个线程同时调用read()方法时，页码计算混乱，出现跳过部分记录的情况，最终导致数据丢失。

修复方案 #

将JdbcPagingItemReader的Bean作用域改为原型（prototype），确保每个线程都能获取到独立的Reader实例，避免状态共享冲突。

修改读取器配置 #

1. 在读取器方法上添加@Scope("prototype")注解：

@Bean
@Scope("prototype") // 关键：设置为原型作用域
public JdbcPagingItemReader<String> tvlTableDataItemReader(DataSource dataSource) throws Exception {
    JdbcPagingItemReader<String> reader = new JdbcPagingItemReader<String>();
    reader.setDataSource(dataSource);
    reader.setPageSize(50);
    reader.setSaveState(true);
    reader.setQueryProvider(createQuery(dataSource));
    reader.setRowMapper(new TVLRowMapper());
    return reader;
}

2. 确保步骤配置中通过Spring容器注入Reader实例：

@Bean
public Step tvlFileCreationStep(JobRepository jobRepository, PlatformTransactionManager transactionManager,
                                JdbcPagingItemReader<String> tvlTableDataItemReader) throws Exception {
    return new StepBuilder("tvl-file-creation-step", jobRepository)
            .<String, String>chunk(50, transactionManager)
            .reader(tvlTableDataItemReader) // 注入原型Bean
            .writer(tvlFileWriter)
            .listener(tvlStepExecutionListener)
            .taskExecutor(tvlTaskExecutor())
            .build();
}

额外验证点 #

• 确认XMLStreamWriter2的线程安全实现是否正确：虽然官方标注线程安全，但需确保写入逻辑中没有额外的非线程安全操作（比如共享的缓冲变量）。
• 若不需要任务重启功能，可尝试关闭saveState（设置reader.setSaveState(false)）辅助验证，不过核心问题仍为Reader的线程安全。

内容的提问来源于stack exchange，提问作者Ashish Shetkar