JSR-352批处理块间延迟问题：无批量删除时如何缩短提交耗时

1. 持久化上下文累积过多实体：Reader标注了@StepScoped，注入的EntityManager默认与Step同生命周期，整个批处理过程中会持有所有已加载（包括删除操作触发级联加载的）实体，导致后续查询时EclipseLink需要维护大量实体的快照、关联关系，开销随块大小线性增长，直接引发延迟。
2. Reader查询无分页逻辑：每次查询都取所有符合条件的前N个ID，数据库需要扫描全表并过滤已删除记录，随着已删除数据增多，查询耗时递增。
3. Writer冗余操作：对find返回的托管实体执行merge操作，额外增加JPA的状态检查开销。

1. 优化EntityManager生命周期与清理 #

通过JSR-352的ChunkListener，在每个块事务提交后清理持久化上下文，释放累积的托管实体：

import javax.batch.api.chunk.listener.AbstractChunkListener;
import javax.inject.Inject;
import javax.inject.Named;
import javax.persistence.EntityManager;
import javax.persistence.PersistenceContext;
import javax.transaction.Status;
import javax.transaction.UserTransaction;

@Named
public class CleanupChunkListener extends AbstractChunkListener {

    @PersistenceContext
    private EntityManager em;

    @Inject
    private UserTransaction utx;

    @Override
    public void afterChunk() throws Exception {
        if (utx.getStatus() == Status.STATUS_COMMITTED) {
            em.clear();
        }
    }
}

在job.xml中为Step配置该监听器：

<step id="deleteEntityStep">
    <chunk reader="myReader" writer="myDeleteWriter" checkpoint-policy="item" item-count="#{chunkSize}">
        <listeners>
            <listener ref="cleanupChunkListener"/>
        </listeners>
    </chunk>
</step>

2. 修改Reader实现分页查询，避免重复扫描已删除数据 #

基于最后处理的ID进行分页查询，利用ID索引加速，避免重复扫描已删除记录：

@StepScoped
@Named
public class MyReader implements AbstractItemReader {

    private int chunkSize = 50;
    
    @PersistenceContext
    private EntityManager em;
    
    private List<Long> currentIdsInChunk;
    private int currentBatchPosition = 0;
    private Long lastProcessedId = 0L;

    // 读取chunkSize的代码省略

    @Override
    public Long readItem() throws Exception {
        if (currentIdsInChunk == null || currentBatchPosition >= currentIdsInChunk.size()) {
            LOGGER.info("Loading Ids for the next chunk...");
            currentIdsInChunk = em.createNativeQuery("SELECT e.id FROM Entity e WHERE e.condition = :condition AND e.id > :lastId ORDER BY e.id")
                             .setParameter("condition", "SOME_CONDITION")
                             .setParameter("lastId", lastProcessedId)
                             .setMaxResults(chunkSize)
                             .getResultList();
            if (currentIdsInChunk.isEmpty()) {
                return null;
            }
            currentBatchPosition = 0;
            lastProcessedId = currentIdsInChunk.get(currentIdsInChunk.size() - 1);
        }
        return currentIdsInChunk.get(currentBatchPosition++);
    }

    @Override
    public Serializable checkpointInfo() {
        return new CheckpointData(lastProcessedId, currentBatchPosition);
    }

    public static class CheckpointData implements Serializable {
        private final Long lastProcessedId;
        private final int currentBatchPosition;

        public CheckpointData(Long lastProcessedId, int currentBatchPosition) {
            this.lastProcessedId = lastProcessedId;
            this.currentBatchPosition = currentBatchPosition;
        }

        public Long getLastProcessedId() { return lastProcessedId; }
        public int getCurrentBatchPosition() { return currentBatchPosition; }
    }

    @Override
    public void open(Serializable checkpoint) throws Exception {
        if (checkpoint instanceof CheckpointData) {
            CheckpointData cp = (CheckpointData) checkpoint;
            this.lastProcessedId = cp.getLastProcessedId();
            this.currentBatchPosition = cp.getCurrentBatchPosition();
            if (currentBatchPosition > 0) {
                currentIdsInChunk = em.createNativeQuery("SELECT e.id FROM Entity e WHERE e.condition = :condition AND e.id > :lastId ORDER BY e.id")
                                 .setParameter("condition", "SOME_CONDITION")
                                 .setParameter("lastId", lastProcessedId - chunkSize)
                                 .setMaxResults(chunkSize)
                                 .getResultList();
            }
        }
    }
}

3. 优化Writer代码，移除冗余操作 #

em.find返回的实体本身就是托管状态，无需执行merge，直接删除即可：

@Named
@Dependent
public class MyDeleteWriter implements NoStateTypedItemWriter<Long> {

    @PersistenceContext
    private EntityManager em;
    
    // constructor省略

    @Override
    public void writeItems(List<Long> ids) throws Exception {
        int deletedCount = 0;
        for (Long id : ids) {
            Entity entity = em.find(Entity.class, id);
            if (entity != null) {
                em.remove(entity);
                deletedCount++;
            } else {
                // 记录实体不存在日志
            }
        }
    }
}

4. 调整EclipseLink缓存设置 #

批量删除操作中，二级缓存可能保留已删除实体快照，增加查询开销。可以关闭实体二级缓存：

@Entity
@Cacheable(false)
public class Entity {
    // 实体字段省略
}

或者在Reader的查询中绕过二级缓存：

currentIdsInChunk = em.createNativeQuery("SELECT e.id FROM Entity e WHERE e.condition = :condition AND e.id > :lastId ORDER BY e.id")
                 .setParameter("condition", "SOME_CONDITION")
                 .setParameter("lastId", lastProcessedId)
                 .setHint("eclipselink.cache-usage", "DoNotCheckCache")
                 .setMaxResults(chunkSize)
                 .getResultList();

5. 数据库层面优化 #

• 创建condition和id的组合索引，加速分页查询：

CREATE INDEX idx_entity_condition_id ON Entity (condition, id);

• 检查级联删除涉及的外键是否有索引，避免删除时全表扫描关联表，减少锁等待时间。

1. 先测试小块大小（如10），确认延迟问题是否缓解。
2. 逐步增大块大小，观察性能变化，找到最优值。
3. 监控数据库锁等待情况，确认是否存在锁冲突导致的延迟。

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