如何使用SpringBoot实现MySQL数据导入MongoDB？

方案一：Spring Data JPA + Spring Data MongoDB 基础批量迁移 #

这是最直接的实现方式，利用Spring Data生态简化数据操作：

1. 添加依赖
   在pom.xml中引入必要依赖：

<!-- MySQL驱动 -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <scope>runtime</scope>
</dependency>
<!-- Spring Data JPA -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>
<!-- Spring Data MongoDB -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-mongodb</artifactId>
</dependency>

2. 配置数据源
   在application.yml中配置MySQL和MongoDB连接：

spring:
  datasource:
    url: jdbc:mysql://localhost:3306/your_mysql_db?useSSL=false&serverTimezone=UTC
    username: root
    password: your_password
    driver-class-name: com.mysql.cj.jdbc.Driver
  jpa:
    hibernate:
      ddl-auto: none
    show-sql: true
  data:
    mongodb:
      uri: mongodb://localhost:27017/your_mongo_db

3. 定义实体类
   分别创建MySQL对应的JPA实体和MongoDB对应的Document实体：

// MySQL实体
@Entity
@Table(name = "user")
public class MysqlUser {
    @Id
    private Long id;
    private String username;
    private String email;
    // 省略getter/setter
}

// MongoDB实体
@Document(collection = "user")
public class MongoUser {
    @Id
    private String id; // Mongo默认用String类型ID，可保留原Long类型或转换
    private String username;
    private String email;
    // 省略getter/setter
}

4. 编写Repository接口

public interface MysqlUserRepository extends JpaRepository<MysqlUser, Long> {}

public interface MongoUserRepository extends MongoRepository<MongoUser, String> {}

5. 实现迁移逻辑
   在Service中批量读取并转换保存：

@Service
public class DataMigrationService {
    private final MysqlUserRepository mysqlRepo;
    private final MongoUserRepository mongoRepo;

    public DataMigrationService(MysqlUserRepository mysqlRepo, MongoUserRepository mongoRepo) {
        this.mysqlRepo = mysqlRepo;
        this.mongoRepo = mongoRepo;
    }

    public void migrateAllUsers() {
        // 批量读取MySQL数据，可分批次避免内存溢出
        List<MysqlUser> mysqlUsers = mysqlRepo.findAll();
        List<MongoUser> mongoUsers = mysqlUsers.stream()
                .map(this::convertToMongoUser)
                .collect(Collectors.toList());
        mongoRepo.saveAll(mongoUsers);
    }

    private MongoUser convertToMongoUser(MysqlUser mysqlUser) {
        MongoUser mongoUser = new MongoUser();
        mongoUser.setId(mysqlUser.getId().toString());
        mongoUser.setUsername(mysqlUser.getUsername());
        mongoUser.setEmail(mysqlUser.getEmail());
        return mongoUser;
    }
}

方案二：Spring Batch 实现可监控的批量迁移 #

适合大规模数据迁移，支持分片处理、失败重试、进度监控：

1. 添加Spring Batch依赖

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-batch</artifactId>
</dependency>

2. 配置Batch元数据存储
   Spring Batch需要元数据库存储任务状态，可复用MySQL：

spring:
  batch:
    jdbc:
      initialize-schema: always # 首次启动初始化表结构

3. 编写Batch组件

• ItemReader（读取MySQL数据）：

@Bean
public JpaPagingItemReader<MysqlUser> mysqlUserReader(EntityManagerFactory entityManagerFactory) {
    return new JpaPagingItemReaderBuilder<MysqlUser>()
            .name("mysqlUserReader")
            .entityManagerFactory(entityManagerFactory)
            .queryString("SELECT u FROM MysqlUser u")
            .pageSize(1000) // 每页读取1000条，避免内存压力
            .build();
}

• ItemProcessor（数据转换）：

@Bean
public ItemProcessor<MysqlUser, MongoUser> userProcessor() {
    return mysqlUser -> {
        MongoUser mongoUser = new MongoUser();
        mongoUser.setId(mysqlUser.getId().toString());
        mongoUser.setUsername(mysqlUser.getUsername());
        mongoUser.setEmail(mysqlUser.getEmail());
        return mongoUser;
    };
}

• ItemWriter（写入MongoDB）：

@Bean
public MongoItemWriter<MongoUser> mongoUserWriter(MongoTemplate mongoTemplate) {
    MongoItemWriter<MongoUser> writer = new MongoItemWriter<>();
    writer.setTemplate(mongoTemplate);
    writer.setCollection("user");
    return writer;
}

4. 配置Job和Step

@Configuration
@EnableBatchProcessing
public class BatchMigrationConfig {
    private final JobBuilderFactory jobBuilderFactory;
    private final StepBuilderFactory stepBuilderFactory;

    public BatchMigrationConfig(JobBuilderFactory jobBuilderFactory, StepBuilderFactory stepBuilderFactory) {
        this.jobBuilderFactory = jobBuilderFactory;
        this.stepBuilderFactory = stepBuilderFactory;
    }

    @Bean
    public Step userMigrationStep(ItemReader<MysqlUser> reader, ItemProcessor<MysqlUser, MongoUser> processor, MongoItemWriter<MongoUser> writer) {
        return stepBuilderFactory.get("userMigrationStep")
                .<MysqlUser, MongoUser>chunk(1000) // 每1000条提交一次
                .reader(reader)
                .processor(processor)
                .writer(writer)
                .build();
    }

    @Bean
    public Job userMigrationJob(Step userMigrationStep) {
        return jobBuilderFactory.get("userMigrationJob")
                .start(userMigrationStep)
                .build();
    }
}

启动SpringBoot应用后，Batch会自动执行迁移任务，可通过Spring Boot Actuator监控任务状态。

方案三：MyBatis + MongoTemplate 自定义迁移 #

如果项目已使用MyBatis，可直接复用Mapper查询数据：

1. 添加MyBatis依赖

<dependency>
    <groupId>org.mybatis.spring.boot</groupId>
    <artifactId>mybatis-spring-boot-starter</artifactId>
    <version>2.2.2</version>
</dependency>

2. 编写MyBatis Mapper

@Mapper
public interface UserMapper {
    List<MysqlUser> listAllUsers();
}

3. 实现迁移逻辑

@Service
public class MyBatisMigrationService {
    private final UserMapper userMapper;
    private final MongoTemplate mongoTemplate;

    public MyBatisMigrationService(UserMapper userMapper, MongoTemplate mongoTemplate) {
        this.userMapper = userMapper;
        this.mongoTemplate = mongoTemplate;
    }

    public void migrateUsers() {
        List<MysqlUser> mysqlUsers = userMapper.listAllUsers();
        List<MongoUser> mongoUsers = mysqlUsers.stream()
                .map(this::convertToMongoUser)
                .collect(Collectors.toList());
        mongoTemplate.insert(mongoUsers, "user");
    }

    private MongoUser convertToMongoUser(MysqlUser mysqlUser) {
        // 转换逻辑同方案一
    }
}

关键注意事项 #

• 批量处理：大规模数据迁移时，务必分批次读取（如按ID分段、分页查询），避免内存溢出。
• 数据映射：注意MySQL和MongoDB的数据类型差异（如日期、ID类型），需在转换时适配。
• 增量迁移：如果需要定时同步增量数据，可添加条件查询（如WHERE update_time > last_migrate_time），配合Spring Schedule实现定时任务。
• 事务处理：MongoDB仅支持单文档事务，批量插入时无需强事务；若需保证数据一致性，可记录迁移日志，失败时重试未迁移的数据。

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