Spring Batch 故障容错中跳过异常记录时重复写入问题的解决方案-java教程-PHP中文网

Spring Batch 故障容错中跳过异常记录时重复写入问题的解决方案

Spring Batch 默认在启用 fault-tolerant 时会对失败项自动重试，若在 Reader/Processor/Writer 中手动捕获异常并写入错误文件，会导致同一错误记录被多次写入；正确做法是使用 SkipListener 统一处理跳过事件，确保每条异常记录仅写入一次。

spring batch 默认在启用 fault-tolerant 时会对失败项自动重试，若在 reader/processor/writer 中手动捕获异常并写入错误文件，会导致同一错误记录被多次写入；正确做法是使用 `skiplistener` 统一处理跳过事件，确保每条异常记录仅写入一次。

在 Spring Batch 中，当配置 .faultTolerant().skipPolicy(...) 后，框架会在处理某条记录抛出可跳过异常时，先重试（默认 3 次），失败后才执行跳过逻辑。但关键在于：跳过（skip）本身是一次性事件，而重试（retry）是独立于 skip 的前置行为。如果你在 ItemReader 或 ItemProcessor 中用 try-catch + FileOutputStream 手动写错误日志，由于重试机制会多次调用该组件，就会导致同一条错误记录被反复写入——这正是你看到“1 条错误输入 → 2 条重复错误输出”的根本原因。

✅ 正确解法：将错误记录写入逻辑完全委托给 SkipListener，它由 Spring Batch 在确认跳过该条目后仅触发一次，严格保证幂等性。

以下为推荐实现方式：

1. 定义线程安全的错误文件写入器（推荐使用 `BufferedWriter` + `FileWriter`）

@Component
public class ErrorRecordWriter {
    private final String errorFilePath = "error_records.log";

    public void writeErrorRecord(Object item, Throwable cause) throws IOException {
        try (BufferedWriter writer = Files.newBufferedWriter(
                Paths.get(errorFilePath), StandardOpenOption.CREATE, StandardOpenOption.APPEND)) {
            String line = String.format("[%s] Item: %s | Cause: %s%n",
                    LocalDateTime.now(), item, cause.toString());
            writer.write(line);
        }
    }
}

2. 实现 `SkipListener` 并注入写入器

@Component
public class CustomSkipListener implements SkipListener<employee employee> {

    private final ErrorRecordWriter errorWriter;

    public CustomSkipListener(ErrorRecordWriter errorWriter) {
        this.errorWriter = errorWriter;
    }

    @Override
    public void onSkipInRead(Throwable throwable) {
        // 若 Reader 抛异常被跳过（如解析失败），此处可记录原始输入行（需 Reader 支持获取当前行号/内容）
    }

    @Override
    public void onSkipInProcess(Employee item, Throwable throwable) {
        try {
            errorWriter.writeErrorRecord(item, throwable);
        } catch (IOException e) {
            throw new RuntimeException("Failed to write skip record to error file", e);
        }
    }

    @Override
    public void onSkipInWrite(Employee item, Throwable throwable) {
        try {
            errorWriter.writeErrorRecord(item, throwable);
        } catch (IOException e) {
            throw new RuntimeException("Failed to write skip record to error file", e);
        }
    }
}</employee>

3. 在 Step 配置中注册 SkipListener（替代手动 try-catch）

@Bean
public Step mainStep(ItemReader<employee> reader,
                     ItemProcessor<employee employee> processor,
                     ItemWriter<employee> writer,
                     CustomSkipListener skipListener) {
    return stepBuilderFactory.get("mainStep")
            .<employee employee>chunk(2)
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .faultTolerant()
                    .skipPolicy(new AlwaysSkipItemSkipPolicy())
                    .skipListener(skipListener) // ✅ 关键：注册监听器
                .end()
            .build();
}</employee></employee></employee></employee>

⚠️ 重要注意事项：

不要在 ItemReader#read()、ItemProcessor#process() 或 ItemWriter#write() 中自行 catch Exception 并写文件——这会与重试/跳过生命周期耦合，破坏框架语义；
SkipListener 的 onSkipInProcess / onSkipInWrite 方法每个被跳过的 item 仅调用一次，无论重试多少次；
若需记录原始文件行内容（如 CSV 解析失败的具体行），应在 ItemReader 中通过 FlatFileItemReader 的 getLineCount() 或自定义 ResourceAwareItemReaderWrapper 捕获上下文，并通过 ThreadLocal 或 StepExecution 属性传递至 SkipListener；
生产环境建议使用异步日志框架（如 Logback 的 AsyncAppender）或消息队列（如 Kafka）替代直接文件 I/O，避免阻塞批处理流程。