Spring Boot3+Java17中SkipListener与Async组件不兼容问题

DevOps 小满

2026-7-10

问题描述

开发Spring Boot应用处理百万级文件记录，采用Spring Batch的AsyncItemProcessor和AsyncItemWriter提升处理效率，但在实现容错逻辑时遇到问题：

处理器执行DB查询获取记录额外数据，希望将DB中不存在的记录写入错误文件，因此使用Spring Batch的SkipListener接口
当查询抛出EmptyResultDataAccessException时，SkipListener未产生任何日志记录
普通同步处理器和写入器的容错逻辑正常工作，仅异步组件存在该问题

询问是否存在配置错误，以及是否有其他方法可在异常发生时将错误记录写入独立文件。

尝试的代码

@Slf4j
@Configuration
@EnableScheduling
public class BatchJob {

    @Scheduled(cron = "${job.batch.cron}")
    public void launchJob() throws Exception {
        Date date = new Date();
        log.debug("In BatchJob - launchJob()");
        log.info("Scheduler starts at [{}]", date);
        JobExecution jobExecution = jobLauncher.run(job(jobRepository, transactionManager), new JobParametersBuilder().addDate("launchDate", date)
                .toJobParameters());
        log.info("Batch job ends with status as [{}]", jobExecution.getStatus());
        log.info("Scheduler ends at [{}]", new Date());
    }

    @Bean
    public Job job(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
        return new JobBuilder("myJob", jobRepository)
                .incrementer(new RunIdIncrementer())
                .listener(jobListener)
                .start(fileValidationStep(jobRepository, transactionManager)).on("FILE_NOT_FOUND").end()
                .from(fileValidationStep(jobRepository, transactionManager)).on("FILE_FOUND").to(recordProcessingStep(jobRepository, transactionManager))
                .from(recordProcessingStep(jobRepository, transactionManager)).on("*").to(fileMoveStep(jobRepository, transactionManager)).end()
                .build();
    }

    @Bean
    public Step fileValidationStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
        return new StepBuilder(fileValidationStep, jobRepository)
                .tasklet(new Tasklet() {
                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {

                        return RepeatStatus.FINISHED;
                    }
                }, transactionManager)
                .build();
    }

    @Bean
    public Step fileMoveStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
        return new StepBuilder("fileMoveStep", jobRepository)
                .tasklet(new Tasklet() {
                    @Override
                    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                        return RepeatStatus.FINISHED;
                    }
                }, transactionManager)
                .build();
    }

    @Bean
    protected Step recordProcessingStep(JobRepository jobRepository, PlatformTransactionManager transactionManager) throws Exception {
        SimpleStepBuilder stepBuilder = new StepBuilder("recordProcessingStep", jobRepository)
                .<DemoRecord, Future<DemoRecord>>chunk(chunkSize, transactionManager)
                .reader(recordReader())
                .processor(asyncProcessor())
                .writer(asyncWriter())
                .faultTolerant()
                .skipLimit(skipLimit)
                .skip(Exception.class)
                .noSkip(FileNotFoundException.class)
                .listener(skipListener)
                .stream(subscriberWriter())
                .stream(dummyWriter());
        stepBuilder.listener(chunkListener);
        return stepBuilder.build();
    }

    @Bean
    @StepScope
    public FlatFileItemReader<DemoRecord> recordReader() {
        FlatFileItemReader<DemoRecord> reader = new FlatFileItemReader<>();
        reader.setResource(new FileSystemResource(inputFilePath));
        reader.setLineMapper(new DefaultLineMapper<DemoRecord>() {{
            setLineTokenizer(new FixedLengthTokenizer() {{
                setNames(inputColumnNames.toArray(new String[0]));
                setColumns(inputColumnRanges);
            }});
            setFieldSetMapper(new BeanWrapperFieldSetMapper<>() {
                {
                    setTargetType(DemoRecord.class);
                }
            });
        }});
        reader.setLinesToSkip(linesToSkip);
        return reader;
    }

    @Bean
    @StepScope
    public ClassifierCompositeItemWriter<DemoRecord> classifierCompositeItemWriter() throws Exception {
        ClassifierCompositeItemWriter<DemoRecord> compositeItemWriter = new ClassifierCompositeItemWriter<>();
        compositeItemWriter.setClassifier(new DemoDataClassifier(subscriberWriter(), dummyWriter()));
        return compositeItemWriter;
    }

    @Bean
    @StepScope
    public FlatFileItemWriter<DemoRecord> subscriberWriter() {
        BeanWrapperFieldExtractor<DemoRecord> fieldExtractor = new BeanWrapperFieldExtractor<>();
        fieldExtractor.setNames(outputColumnNames);
        fieldExtractor.afterPropertiesSet();

        FormatterLineAggregator<DemoRecord> lineAggregator = new FormatterLineAggregator<>();
        lineAggregator.setFormat(outputColumnFormats);
        lineAggregator.setFieldExtractor(fieldExtractor);
        return new FlatFileItemWriterBuilder<DemoRecord>()
                .name("subscriberWriter")
                .resource(new FileSystemResource(targetDir))
                .lineAggregator(lineAggregator)
                .shouldDeleteIfExists(true)
                .shouldDeleteIfEmpty(true)
                .build();
    }

    @Bean
    @StepScope
    public FlatFileItemWriter<DemoRecord> dummyWriter() {
        BeanWrapperFieldExtractor<DemoRecord> fieldExtractor = new BeanWrapperFieldExtractor<>();
        fieldExtractor.setNames(outputColumnNames);
        fieldExtractor.afterPropertiesSet();

        FormatterLineAggregator<DemoRecord> lineAggregator = new FormatterLineAggregator<>();
        lineAggregator.setFormat(outputColumnFormats);
        lineAggregator.setFieldExtractor(fieldExtractor);
        return new FlatFileItemWriterBuilder<DemoRecord>()
                .name("dummyWriter")
                .resource(new FileSystemResource(target))
                .lineAggregator(lineAggregator)
                .shouldDeleteIfExists(true)
                .shouldDeleteIfEmpty(true)
                .build();
    }

    @Bean
    @StepScope
    public AsyncItemProcessor<DemoRecord, DemoRecord> asyncProcessor() {
        AsyncItemProcessor<DemoRecord, DemoRecord> asyncItemProcessor = new AsyncItemProcessor<>();
        asyncItemProcessor.setDelegate(recordProcessor);
        asyncItemProcessor.setTaskExecutor(taskExecutor());
        return asyncItemProcessor;
    }

    @Bean
    @StepScope
    public AsyncItemWriter<DemoRecord> asyncWriter() throws Exception {
        AsyncItemWriter<DemoRecord> asyncItemWriter = new AsyncItemWriter<>();
        asyncItemWriter.setDelegate(classifierCompositeItemWriter());
        return asyncItemWriter;
    }

    @Bean
    public TaskExecutor taskExecutor() {
        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
        executor.setCorePoolSize(64);
        executor.setMaxPoolSize(64);
        executor.setQueueCapacity(64);
        executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
        executor.setThreadNamePrefix("MultiThreaded-");
        return executor;
    }

}

（注：修正了代码中几处语法错误，如资源路径的括号、列名数组转换的语法）

解决方案分析

核心问题原因

AsyncItemProcessor会将处理器的执行逻辑提交到异步线程池，执行结果包装在Future对象中。当委托处理器抛出EmptyResultDataAccessException时，异常会被封装到Future内部，而非直接抛出到主线程的步骤容错逻辑中。只有当AsyncItemWriter调用Future.get()时，异常才会以ExecutionException的形式抛出，但默认配置下，Spring Batch的容错机制可能未正确捕获并触发SkipListener。

配置调整方案

明确捕获包装后的异常
调整步骤的skip配置，增加对ExecutionException的支持，同时在SkipListener中解析该异常获取原始的EmptyResultDataAccessException：

.skipLimit(skipLimit)
.skip(Exception.class)
.skip(ExecutionException.class)
.noSkip(FileNotFoundException.class)

在SkipListener的onSkipInProcess方法中解析异常：

@Override
public void onSkipInProcess(DemoRecord item, Throwable t) {
    if (t instanceof ExecutionException && t.getCause() instanceof EmptyResultDataAccessException) {
        // 记录错误日志并处理错误记录
        log.error("Record {} not found in DB", item.getId(), t.getCause());
        // 可直接调用错误文件写入器写入该记录
    }
}

确保AsyncItemWriter正确传递异常
确认AsyncItemWriter在处理Future结果时未捕获异常并吞掉，默认实现会在Future.get()时抛出异常，因此无需额外修改，但需确保委托写入器未屏蔽异常。

替代实现方案（无需依赖SkipListener）

直接在业务逻辑中标记错误记录，通过ClassifierCompositeItemWriter路由到错误文件：

修改委托处理器recordProcessor，捕获EmptyResultDataAccessException，将记录标记为错误状态（比如新增一个isError字段）：

public DemoRecord process(DemoRecord item) {
    try {
        // DB查询逻辑
        ExtraData data = dbRepository.findByItemId(item.getId());
        item.setExtraData(data);
        item.setIsError(false);
    } catch (EmptyResultDataAccessException e) {
        item.setIsError(true);
    }
    return item;
}

修改DemoDataClassifier，根据isError字段将记录路由到错误文件写入器（比如errorWriter，需新增该写入器Bean）：

public class DemoDataClassifier implements Classifier<DemoRecord, ItemWriter<? super DemoRecord>> {
    private final ItemWriter<DemoRecord> successWriter;
    private final ItemWriter<DemoRecord> errorWriter;

    public DemoDataClassifier(ItemWriter<DemoRecord> successWriter, ItemWriter<DemoRecord> errorWriter) {
        this.successWriter = successWriter;
        this.errorWriter = errorWriter;
    }

    @Override
    public ItemWriter<? super DemoRecord> classify(DemoRecord item) {
        return item.isError() ? errorWriter : successWriter;
    }
}

更新classifierCompositeItemWriter的配置，传入错误写入器，同时在步骤中移除faultTolerant和SkipListener相关配置（如果不需要其他跳过逻辑）。

注意事项

异步线程池的核心线程数设置为64需结合服务器实际CPU核心数调整，避免线程过多导致上下文切换开销过大。
使用异步组件时，需确保所有依赖的Bean（如数据库连接池）支持多线程访问。

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

Spring Boot3+Java17中SkipListener与Async组件不兼容问题

核心问题原因 #

配置调整方案 #

替代实现方案（无需依赖SkipListener） #

注意事项 #

核心问题原因

配置调整方案

替代实现方案（无需依赖SkipListener）

注意事项