Spring Batch迁移MongoDB数据时如何正确更新现有项？

你的问题核心是每批处理都重复执行全局删除，导致前一批插入的数据被后一批删除。用Tasklet可以把删除操作移到整个迁移流程的最开始，只执行一次，彻底解决数据丢失问题，具体实现如下：

核心思路 #

将目标集合的匹配数据删除操作，从每批读写的Chunk逻辑中抽离，用Tasklet作为独立的前置Step，在整个迁移Job启动时仅执行一次，之后的批量读取和写入不再触发删除，避免覆盖已插入的数据。

具体实现步骤 #

1. 编写前置清理Tasklet #

创建一个Tasklet组件，负责一次性删除目标集合中符合迁移条件的所有文档：

@Component
public class PreMigrationCleanupTasklet implements Tasklet {

    private final MongoTemplate mongoTemplate;

    // 构造注入MongoTemplate
    public PreMigrationCleanupTasklet(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
    }

    @Override
    public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
        // 替换为你的迁移过滤条件
        Query deleteQuery = new Query(Criteria.where("yourFilterField").is("targetFilterValue"));
        // 删除目标集合中匹配的文档
        mongoTemplate.remove(deleteQuery, TargetCollectionEntity.class);
        return RepeatStatus.FINISHED;
    }
}

2. 配置Job执行流程 #

把清理Tasklet作为第一个Step，之后再执行原来的Chunk式读写Step，保证删除操作仅执行一次：

@Configuration
public class MigrationJobConfig {

    @Autowired
    private JobBuilderFactory jobBuilderFactory;

    @Autowired
    private StepBuilderFactory stepBuilderFactory;

    @Autowired
    private PreMigrationCleanupTasklet cleanupTasklet;

    @Autowired
    private MongoCursorItemReader<SourceEntity> sourceMongoReader;

    @Autowired
    private ItemWriter<TargetEntity> targetMongoWriter;

    // 前置清理Step
    @Bean
    public Step preCleanupStep() {
        return stepBuilderFactory.get("preCleanupStep")
                .tasklet(cleanupTasklet)
                .build();
    }

    // 批量迁移Step（原有的读写逻辑）
    @Bean
    public Step batchMigrationStep() {
        return stepBuilderFactory.get("batchMigrationStep")
                .<SourceEntity, TargetEntity>chunk(1000) // 你的批处理大小
                .reader(sourceMongoReader)
                .processor(/* 如果需要数据转换，这里添加你的ItemProcessor */)
                .writer(targetMongoWriter)
                .build();
    }

    // 组装整个迁移Job
    @Bean
    public Job mongoCollectionMigrationJob() {
        return jobBuilderFactory.get("mongoCollectionMigrationJob")
                .start(preCleanupStep())
                .next(batchMigrationStep())
                .build();
    }
}

额外优化方案 #

如果你的场景不需要全量覆盖目标数据，也可以跳过删除步骤，改用Upsert操作避免重复键错误：
在ItemWriter中使用mongoTemplate.upsert()代替insert()，实现“存在则更新、不存在则插入”的逻辑，既避免重复键异常，也无需删除数据：

public class UpsertTargetWriter implements ItemWriter<TargetEntity> {

    private final MongoTemplate mongoTemplate;

    public UpsertTargetWriter(MongoTemplate mongoTemplate) {
        this.mongoTemplate = mongoTemplate;
    }

    @Override
    public void write(List<? extends TargetEntity> items) throws Exception {
        for (TargetEntity item : items) {
            // 根据唯一键构建查询条件
            Query query = new Query(Criteria.where("uniqueKeyField").is(item.getUniqueKeyField()));
            Update update = Update.fromObject(item);
            // 执行Upsert
            mongoTemplate.upsert(query, update, TargetEntity.class);
        }
    }
}

内容的提问来源于stack exchange，提问作者Murat Yıldız