ItemStream 与状态管理

本章定位：深入掌握 ItemStream 接口——Reader/Writer 的资源和状态管理器，理解 open/update/close 生命周期和 ExecutionContext 驱动的断点续传机制。

定义与作用

ItemStream 是 Spring Batch 中用于管理资源和持久化状态的接口。实现了 ItemStream 的 Reader/Writer 可以在 Step 执行期间安全地打开/关闭资源，并在每个 Chunk 提交后将进度写入 ExecutionContext。

public interface ItemStream {
    void open(ExecutionContext executionContext) throws ItemStreamException;
    void update(ExecutionContext executionContext) throws ItemStreamException;
    void close() throws ItemStreamException;
}

飞翔科技架构师白歌的比喻：

"ItemStream 是工人手中的签到簿。open() 翻开签到簿看看上次做到哪了，update() 做完一批活记下进度，close() 下班合上签到簿。如果中途停电，明天翻开就知道从哪继续。"

核心原理

三个方法的调用时机

方法	调用时机	典型操作
`open()`	Step 启动时（首次或重启）	打开文件/连接，从 ExecutionContext 恢复进度
`update()`	每个 Chunk 提交后	将当前进度写入 ExecutionContext
`close()`	Step 结束时（正常或异常）	关闭文件/连接

完整示例

场景一：自定义文件 Reader 实现断点续传

public class ResumableFileReader implements ItemReader<String>, ItemStream {

    private final String filePath;
    private BufferedReader reader;
    private int currentLine = 0;
    private int lastCommittedLine = 0;

    public ResumableFileReader(String filePath) {
        this.filePath = filePath;
    }

    @Override
    public void open(ExecutionContext executionContext) {
        // 1. 恢复上次进度的行号
        if (executionContext.containsKey("current.line")) {
            this.currentLine = executionContext.getInt("current.line");
            this.lastCommittedLine = this.currentLine;
            System.out.println("Restart: 从第 " + (currentLine + 1) + " 行继续");
        }

        // 2. 打开文件并跳过已处理的行
        try {
            this.reader = new BufferedReader(new FileReader(filePath));
            for (int i = 0; i < currentLine; i++) {
                reader.readLine(); // 跳过已处理的行
            }
        } catch (IOException e) {
            throw new ItemStreamException("Failed to open file", e);
        }
    }

    @Override
    public String read() throws Exception {
        if (reader == null) return null;
        String line = reader.readLine();
        if (line != null) {
            currentLine++;
        }
        return line;
    }

    @Override
    public void update(ExecutionContext executionContext) {
        // 3. 每个 Chunk 提交后保存当前进度
        lastCommittedLine = currentLine;
        executionContext.putInt("current.line", currentLine);
    }

    @Override
    public void close() {
        try {
            if (reader != null) reader.close();
        } catch (IOException ignored) {}
    }
}

场景二：Step 配置中使用自定义 Reader

@Bean
public Step resumableStep(JobRepository jobRepository,
                            PlatformTransactionManager tx) {
    return new StepBuilder("resumableStep", jobRepository)
            .<String, String>chunk(50, tx)
            .reader(new ResumableFileReader("data/orders.txt"))
            .writer(items -> {
                for (String item : items) {
                    if (item.contains("CRASH_AFTER_THIS")) {
                        throw new RuntimeException("Simulated crash!");
                    }
                    System.out.println("Processed: " + item);
                }
            })
            .build();
}

操作前后对比：

首次执行:
  Chunk 1: 50 行 → COMMIT → ExecutionContext: current.line=50
  Chunk 2: 50 行 → COMMIT → ExecutionContext: current.line=100
  Chunk 3: 读到第 123 行 "CRASH_AFTER_THIS" → write() 抛异常 → ROLLBACK
  JobExecution: FAILED | ExecutionContext 中 current.line=100

重启执行（相同参数）:
  open(): 从 ExecutionContext 恢复 current.line=100
  reader 跳过前 100 行，从第 101 行开始
  Chunk 1: 第 101-150 行 → 正常处理 → COMMIT
  ✅ 前 100 行不会重复处理

易错场景与避坑

反例一：忘记在 open() 中恢复状态

// ❌ open() 中忽略 ExecutionContext
@Override
public void open(ExecutionContext executionContext) {
    // 不读取执行上下文 → 重启时丢失进度
    this.reader = new BufferedReader(new FileReader(filePath));
    // 每次重启都从第 0 行开始 → 前面已处理的数据被重复处理！
}

反例二：update() 在 read() 内部而非单独调用

// ❌ 每条 read 都更新 ExecutionContext → 性能灾难
@Override
public String read() {
    String line = reader.readLine();
    if (line != null) {
        currentLine++;
        // ❌ 不要在这里更新！每条数据写一次 DB 太慢
    }
    return line;
}

正确做法：update() 只由框架在每个 Chunk 提交后调用一次。

面试高频考点

Q1：ItemStream 的 update() 和 open() 是如何配合的？

update() 在每个 Chunk COMMIT 后被框架调用，将当前进度写入 ExecutionContext。open() 在 Step 启动时被调用，从中恢复进度。两者之间框架负责将 ExecutionContext 持久化到 BATCH_STEP_EXECUTION_CONTEXT 表。

Q2：如果 Reader 没有实现 ItemStream，重启时会怎样？

框架不会自动管理该 Reader 的状态。重启时 Reader 从头开始读取，可能导致数据重复处理。因此任何需要断点续传的 Reader/Writer 都必须实现 ItemStream。

上一章：ExecutionContext 详解下一章：ItemReader 详解