Spring Batch CSV转DB场景异常处理及统一错误CSV生成方案咨询

原帖的核心解法是基于Spring Batch的SkipListener机制捕获各阶段跳过的异常记录，你可以通过扩展该监听器收集错误字段并写入统一错误CSV文件，适配你的需求的完整实现如下：

所有异常分为读取、处理、写入三个阶段，分别对应你提到的CSV格式错误、业务校验错误、DB写入错误三类场景，通过SkipListener统一捕获三类异常，组装要求的错误字段后写入错误文件。

1. 基础实体定义 #

首先定义错误记录实体、CSV读取实体，存储需要的元数据：

// 错误记录实体，对应错误CSV的字段
public class CsvErrorRecord {
    // 原始CSV行内容
    private String originalLine;
    // 异常原因
    private String errorReason;
    // 所属文件名
    private String fileName;
    // 行号
    private Long lineNumber;

    // 构造方法、getter、setter省略
}

// CSV读取实体，额外存储元数据用于异常收集
public class CsvRecord {
    // 业务字段省略
    private String originalLine;
    private Long lineNumber;

    // 构造方法、getter、setter省略
}

2. 自定义LineMapper，存原始行和行号 #

读取CSV时就把原始行内容、行号存到读取实体中，供后续异常场景使用：

public class CsvLineMapper extends DefaultLineMapper<CsvRecord> {
    @Override
    public CsvRecord mapLine(String line, int lineNumber) throws Exception {
        CsvRecord record = super.mapLine(line, lineNumber);
        record.setOriginalLine(line);
        record.setLineNumber((long) lineNumber);
        return record;
    }
}

3. 自定义SkipListener，统一捕获异常 #

public class CsvSkipListener implements SkipListener<CsvRecord, Object> {
    private final ItemWriter<CsvErrorRecord> errorWriter;
    private final String fileName;

    // 构造方法注入错误文件写入器、当前处理的CSV文件名
    public CsvSkipListener(ItemWriter<CsvErrorRecord> errorWriter, String fileName) {
        this.errorWriter = errorWriter;
        this.fileName = fileName;
    }

    // 处理读取阶段异常：对应CSV格式错误场景
    @Override
    public void onSkipInRead(Throwable t) {
        if (t instanceof FlatFileParseException parseEx) {
            CsvErrorRecord error = new CsvErrorRecord();
            error.setOriginalLine(parseEx.getInput());
            error.setLineNumber((long) parseEx.getLineNumber());
            error.setFileName(fileName);
            error.setErrorReason("CSV格式错误：" + t.getMessage());
            writeError(error);
        }
    }

    // 处理处理阶段异常：对应业务校验错误场景
    @Override
    public void onSkipInProcess(CsvRecord item, Throwable t) {
        CsvErrorRecord error = new CsvErrorRecord();
        error.setOriginalLine(item.getOriginalLine());
        error.setLineNumber(item.getLineNumber());
        error.setFileName(fileName);
        error.setErrorReason("业务校验错误：" + t.getMessage());
        writeError(error);
    }

    // 处理写入阶段异常：对应DB写入错误场景
    @Override
    public void onSkipInWrite(Object item, Throwable t) {
        // 若写入实体没有存原始行元数据，可在转换时把CsvRecord的元数据同步到写入实体
        CsvRecord record = (CsvRecord) item; // 可根据实际写入实体类型调整转换逻辑
        CsvErrorRecord error = new CsvErrorRecord();
        error.setOriginalLine(record.getOriginalLine());
        error.setLineNumber(record.getLineNumber());
        error.setFileName(fileName);
        error.setErrorReason("DB写入错误：" + t.getMessage());
        writeError(error);
    }

    private void writeError(CsvErrorRecord error) {
        try {
            errorWriter.write(List.of(error));
        } catch (Exception e) {
            // 错误文件写入失败时可打印本地日志兜底
            e.printStackTrace();
        }
    }
}

4. 错误CSV写入器配置 #

@Bean
public ItemWriter<CsvErrorRecord> errorCsvWriter() {
    FlatFileItemWriter<CsvErrorRecord> writer = new FlatFileItemWriter<>();
    // 错误文件输出路径
    writer.setResource(new FileSystemResource("csv_error_output.csv"));
    // 写入CSV表头
    writer.setHeaderCallback(w -> w.write("原始行内容,异常原因,所属文件名,行号"));
    // 配置字段映射
    DelimitedLineAggregator<CsvErrorRecord> aggregator = new DelimitedLineAggregator<>();
    aggregator.setDelimiter(",");
    BeanWrapperFieldExtractor<CsvErrorRecord> extractor = new BeanWrapperFieldExtractor<>();
    extractor.setNames(new String[]{"originalLine", "errorReason", "fileName", "lineNumber"});
    aggregator.setFieldExtractor(extractor);
    writer.setLineAggregator(aggregator);
    // 初始化写入器
    try {
        writer.afterPropertiesSet();
    } catch (Exception e) {
        throw new RuntimeException(e);
    }
    return writer;
}

5. Step容错配置 #

@Bean
public Step csvProcessStep(StepBuilderFactory stepBuilderFactory,
                           ItemReader<CsvRecord> csvReader,
                           ItemProcessor<CsvRecord, Object> csvProcessor,
                           ItemWriter<Object> dbWriter,
                           ItemWriter<CsvErrorRecord> errorCsvWriter) {
    return stepBuilderFactory.get("csvProcessStep")
            .chunk(10)
            .reader(csvReader)
            .processor(csvProcessor)
            .writer(dbWriter)
            // 开启容错配置
            .faultTolerant()
            // 配置需要跳过的异常，可按需指定具体异常类替代通用Exception
            .skip(Exception.class)
            // 跳过次数上限，设置为最大值即可跳过所有错误行
            .skipLimit(Integer.MAX_VALUE)
            // 注册自定义跳过监听器
            .listener(new CsvSkipListener(errorCsvWriter, "input.csv")) // 替换为实际文件名
            .build();
}

• 如果原始CSV行包含逗号等特殊字符，可配置错误CSV写入器用双引号包裹字段，避免格式错乱
• 跳过规则可按需细化，比如仅跳过FlatFileParseException、自定义业务异常、DataIntegrityViolationException三类对应你的场景，避免跳过不可控异常
• 写入阶段如果使用的是独立的实体类而非读取用的CsvRecord，需要在转换时把原始行、行号等元数据同步到写入实体中

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