如何在ItemReader锁定行时实现ItemWriter向数据库写入数据？

我定义了一个简单的Spring Batch Job，流程为从数据库读取记录、处理数据项，最后将处理成功的数据项的status列更新。但当ItemWriter尝试提交事务写入chunk时出现卡顿，原因是ItemReader已锁定表中数据。ItemReader的查询语句如下：

SELECT ... FOR UPDATE SKIP LOCKED LIMIT 500

我已尝试定义自定义PlatformTransactionManager Bean，代码如下：

@Bean
public PlatformTransactionManager transactionManager(final DataSource dataSource) {
    return new JdbcTransactionManager(dataSource);
}

请问如何在ItemReader锁定行的前提下，让ItemWriter能够正常向数据库写入数据？

1. 拆分读写事务（两步事务策略） #

Spring Batch默认会将ItemReader、Processor、Writer纳入同一个事务，导致Reader锁定的行在整个chunk处理周期内持续被持有，Writer提交时必然出现锁冲突。你可以配置两步事务策略，让Reader在独立事务中执行，完成后立即释放锁，再在单独事务中完成数据处理与写入：

@Bean
public Step myStep(StepBuilderFactory stepBuilderFactory,
                   ItemReader<MyEntity> itemReader,
                   ItemProcessor<MyEntity, MyEntity> itemProcessor,
                   ItemWriter<MyEntity> itemWriter) {
    return stepBuilderFactory.get("myStep")
            .<MyEntity, MyEntity>chunk(500)
            .reader(itemReader)
            .processor(itemProcessor)
            .writer(itemWriter)
            .transactionAttribute(getTransactionAttribute())
            .build();
}

private DefaultTransactionAttribute getTransactionAttribute() {
    DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
    // 设置Reader的事务传播行为为REQUIRES_NEW，确保开启独立事务
    attribute.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    return attribute;
}

该方案下，Reader的事务在读取完成后立即提交释放锁，Writer处理的是已加载到内存的数据，更新操作不会被Reader的锁阻塞。

2. 自定义事务型Reader控制锁时机 #

通过自定义ItemReader，将读取操作封装在独立短事务中，读取完成后立即提交释放锁，避免锁长时间持有：

public class TransactionalItemReader implements ItemReader<MyEntity> {
    private final JdbcTemplate jdbcTemplate;
    private final PlatformTransactionManager transactionManager;
    private List<MyEntity> currentBatch;
    private int batchIndex = 0;

    public TransactionalItemReader(JdbcTemplate jdbcTemplate, PlatformTransactionManager transactionManager) {
        this.jdbcTemplate = jdbcTemplate;
        this.transactionManager = transactionManager;
    }

    @Override
    public MyEntity read() throws Exception {
        if (currentBatch == null || batchIndex >= currentBatch.size()) {
            // 开启独立事务读取数据
            TransactionStatus status = transactionManager.getTransaction(new DefaultTransactionAttribute());
            try {
                currentBatch = jdbcTemplate.query("SELECT ... FOR UPDATE SKIP LOCKED LIMIT 500",
                        (rs, rowNum) -> {
                            // 实现ResultSet到MyEntity的映射逻辑
                            MyEntity entity = new MyEntity();
                            entity.setId(rs.getLong("id"));
                            entity.setStatus(rs.getString("status"));
                            // 其他字段映射
                            return entity;
                        });
                transactionManager.commit(status);
                batchIndex = 0;
            } catch (Exception e) {
                transactionManager.rollback(status);
                throw e;
            }
        }
        return batchIndex < currentBatch.size() ? currentBatch.get(batchIndex++) : null;
    }
}

3. 调整数据库事务隔离级别 #

过高的隔离级别会加剧锁竞争，建议将默认隔离级别调整为READ_COMMITTED（多数业务场景的合理选择），兼顾一致性与并发性能：

@Bean
public PlatformTransactionManager transactionManager(final DataSource dataSource) {
    JdbcTransactionManager transactionManager = new JdbcTransactionManager(dataSource);
    // 设置默认隔离级别为READ_COMMITTED
    transactionManager.setDefaultTransactionIsolationLevel(Connection.TRANSACTION_READ_COMMITTED);
    return transactionManager;
}

同时确认你的数据库支持SKIP LOCKED语法（如PostgreSQL 9.5+、MySQL 8.0+等），旧版本数据库的锁机制可能存在差异。

4. 优化Chunk大小 #

当前Chunk设置为500，若单批次数据处理耗时较长，锁持有时间会相应增加。可适当减小Chunk大小（如100或200），缩短锁的持有周期，降低Writer的等待概率。

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