JSR-352批量作业：如何让ItemReader读取后立即保存Chunk checkpoint？

针对你遇到的JSR-352批处理场景——数据源无法在并行环境重复获取同一Chunk、需实现可复用的ItemReader适配重试需求，且不想依赖额外监听器或 transient 数据，可通过以下设计实现：

核心思路 #

将Chunk的预加载行为与Checkpoint强绑定：首次读取时一次性拉取整个Chunk的数据并持久化，同时在Checkpoint中记录该Chunk的状态（已预加载、已读取位置）；当Chunk重试时，通过Checkpoint判断已预加载过该Chunk，直接复用持久化的数据，无需再次访问数据源。

具体实现步骤 #

1. 定义Checkpoint数据结构 #

创建可序列化的Checkpoint类，存储Chunk关键状态：

public class PreloadedChunkCheckpoint implements Serializable {
    // 唯一标识当前Chunk（结合分区ID、时间戳生成）
    private String chunkId;
    // 当前已读取的数据项索引
    private int readIndex;
    // 标记是否已预加载当前Chunk的数据
    private boolean isChunkPreloaded;

    // 构造器、getter/setter 省略
}

2. 实现通用ItemReader核心逻辑 #

public class PreloadingItemReader<T> implements ItemReader<T>, CheckpointInfo {
    // 注入数据源操作类，用于拉取Chunk数据
    @Inject
    private DataSourceFetcher<T> dataSourceFetcher;
    // 注入分区上下文，用于生成唯一ChunkID
    @Inject
    private PartitionContext partitionContext;

    // 预加载的Chunk数据缓存
    private Queue<T> chunkDataQueue;
    // 当前Checkpoint实例
    private PreloadedChunkCheckpoint checkpoint;

    @Override
    public void open(Serializable checkpoint) throws Exception {
        if (checkpoint instanceof PreloadedChunkCheckpoint) {
            this.checkpoint = (PreloadedChunkCheckpoint) checkpoint;
            // 若Checkpoint标记已预加载，从持久化存储恢复数据
            if (this.checkpoint.isChunkPreloaded()) {
                this.chunkDataQueue = restoreChunkData(this.checkpoint.getChunkId());
            }
        } else {
            // 首次启动，初始化空Checkpoint
            this.checkpoint = new PreloadedChunkCheckpoint();
            this.checkpoint.setChunkId(generateChunkId());
            this.checkpoint.setReadIndex(0);
            this.checkpoint.setChunkPreloaded(false);
        }
    }

    @Override
    public T readItem() throws Exception {
        // 首次读取时预加载整个Chunk
        if (!checkpoint.isChunkPreloaded()) {
            // 一次性拉取当前Chunk的所有数据
            this.chunkDataQueue = dataSourceFetcher.fetchEntireChunk(partitionContext);
            // 持久化预加载数据（示例为本地临时文件，可替换为数据库临时表/分布式缓存）
            persistChunkData(checkpoint.getChunkId(), chunkDataQueue);
            // 更新Checkpoint状态
            checkpoint.setChunkPreloaded(true);
            checkpoint.setReadIndex(0);
        }

        // 从缓存读取下一个数据项
        T item = chunkDataQueue.poll();
        if (item != null) {
            checkpoint.setReadIndex(checkpoint.getReadIndex() + 1);
        }
        return item;
    }

    @Override
    public Serializable checkpointInfo() throws Exception {
        // 返回当前Checkpoint，由容器负责保存
        return this.checkpoint;
    }

    @Override
    public void close() throws Exception {
        // Chunk处理成功后清理临时存储的数据
        if (checkpoint.isChunkPreloaded()) {
            cleanupChunkData(checkpoint.getChunkId());
        }
    }

    // 生成唯一ChunkID，避免跨分区冲突
    private String generateChunkId() {
        return partitionContext.getPartitionId() + "_" + System.currentTimeMillis();
    }

    // 持久化Chunk数据到临时文件
    private void persistChunkData(String chunkId, Queue<T> data) throws IOException {
        File tempFile = new File(System.getProperty("java.io.tmpdir"), chunkId + ".dat");
        try (ObjectOutputStream oos = new ObjectOutputStream(new FileOutputStream(tempFile))) {
            oos.writeObject(data);
        }
    }

    // 从临时文件恢复Chunk数据
    private Queue<T> restoreChunkData(String chunkId) throws IOException, ClassNotFoundException {
        File tempFile = new File(System.getProperty("java.io.tmpdir"), chunkId + ".dat");
        try (ObjectInputStream ois = new ObjectInputStream(new FileInputStream(tempFile))) {
            return (Queue<T>) ois.readObject();
        }
    }

    // 清理临时文件
    private void cleanupChunkData(String chunkId) {
        File tempFile = new File(System.getProperty("java.io.tmpdir"), chunkId + ".dat");
        if (tempFile.exists()) {
            tempFile.delete();
        }
    }
}

3. 关键细节说明 #

• 持久化可靠性：重试时Reader实例可能被重建，内存缓存不可靠，因此将预加载数据序列化到临时存储（本地文件/数据库），确保重试时能恢复。
• Checkpoint状态同步：首次预加载完成后立即更新Checkpoint标记，容器会在后续的Checkpoint保存时机保留该状态，重试时直接复用缓存数据，无需再次访问数据源。
• 分区安全：每个分区的Reader使用独立的ChunkID，不会出现跨分区的数据冲突，符合并行处理要求。

极简替代方案：依赖作业重启机制 #

如果不想实现复杂的预加载逻辑，可直接利用数据源仅允许作业重启时重复获取Chunk的特性：

• 在Reader中记录当前处理的Chunk标识，当发生可重试异常时让作业直接失败。
• 重启作业时，Reader根据Checkpoint中的Chunk标识，重新从数据源获取该Chunk（作业重启为单实例初始化，数据源允许此时重复获取）。

此方案缺点是作业中断后需手动/自动重启，不如预加载方案平滑。

内容的提问来源于stack exchange，提问作者Андрей Андреев