[定時(shí)觸發(fā)器] -> [數(shù)據(jù)抽取] -> [數(shù)據(jù)轉(zhuǎn)換] -> [數(shù)據(jù)加載] -> [結(jié)果通知]

<!-- Spring Boot Starter Web (包含Scheduler) -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-web</artifactId>
</dependency>

<!-- Spring Data JPA -->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-data-jpa</artifactId>
</dependency>

<!-- 數(shù)據(jù)庫(kù)驅(qū)動(dòng)（示例使用MySQL） -->
<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.28</version>
</dependency>

@SpringBootApplication
@EnableScheduling
public class DataSyncApplication {
    public static void main(String[] args) {
        SpringApplication.run(DataSyncApplication.class, args);
    }
}

@Component
public class DataSyncScheduler {
    
    private static final Logger logger = LoggerFactory.getLogger(DataSyncScheduler.class);
    
    @Autowired
    private SourceRepository sourceRepo;
    
    @Autowired
    private TargetRepository targetRepo;

    // 每天凌晨1點(diǎn)執(zhí)行
    @Scheduled(cron = "0 0 1 * * ?")
    @Transactional(propagation = Propagation.REQUIRED, isolation = Isolation.READ_COMMITTED)
    public void syncDataDaily() {
        try {
            logger.info("開始數(shù)據(jù)同步任務(wù)...");
            
            // 1. 獲取增量數(shù)據(jù)
            LocalDateTime lastSyncTime = getLastSyncTime();
            List<SourceEntity> newData = sourceRepo.findByUpdateTimeAfter(lastSyncTime);
            
            // 2. 數(shù)據(jù)轉(zhuǎn)換
            List<TargetEntity> transformedData = transformData(newData);
            
            // 3. 批量保存
            targetRepo.saveAll(transformedData);
            
            // 4. 更新同步時(shí)間
            updateLastSyncTime(LocalDateTime.now());
            
            logger.info("數(shù)據(jù)同步完成，處理記錄數(shù)：{}", newData.size());
        } catch (Exception e) {
            logger.error("數(shù)據(jù)同步任務(wù)異常：", e);
            // 添加重試或報(bào)警邏輯
        }
    }
    
    // 數(shù)據(jù)轉(zhuǎn)換方法
    private List<TargetEntity> transformData(List<SourceEntity> sourceList) {
        return sourceList.stream()
                .map(entity -> new TargetEntity(
                        entity.getId(),
                        entity.getName(),
                        entity.getData(),
                        LocalDateTime.now()))
                .collect(Collectors.toList());
    }
    
    // 獲取上次同步時(shí)間（示例）
    private LocalDateTime getLastSyncTime() {
        // 可從數(shù)據(jù)庫(kù)或緩存獲取
        return LocalDateTime.now().minusDays(1);
    }
    
    // 更新同步時(shí)間
    private void updateLastSyncTime(LocalDateTime time) {
        // 持久化存儲(chǔ)邏輯
    }
}

// 使用Redis實(shí)現(xiàn)分布式鎖
@Scheduled(cron = "${sync.cron}")
public void distributedSyncTask() {
    String lockKey = "data_sync_lock";
    String requestId = UUID.randomUUID().toString();
    try {
        // 嘗試獲取鎖
        boolean locked = redisTemplate.opsForValue()
                .setIfAbsent(lockKey, requestId, 30, TimeUnit.MINUTES);
        
        if (locked) {
            // 執(zhí)行同步邏輯
            performSync();
        }
    } finally {
        // 釋放鎖
        if (requestId.equals(redisTemplate.opsForValue().get(lockKey))) {
            redisTemplate.delete(lockKey);
        }
    }
}

private void batchSyncWithPagination() {
    int pageSize = 1000;
    int page = 0;
    
    Page<SourceEntity> dataPage;
    do {
        dataPage = sourceRepo.findAll(PageRequest.of(page, pageSize));
        List<TargetEntity> targetList = transformData(dataPage.getContent());
        targetRepo.saveAll(targetList);
        page++;
    } while (dataPage.hasNext());
}

# application.yml
spring:
  jpa:
    properties:
      hibernate:
        jdbc:
          batch_size: 500
        order_inserts: true
        order_updates: true

@Aspect
@Component
public class SyncMonitorAspect {
    
    @Around("@annotation(org.springframework.scheduling.annotation.Scheduled)")
    public Object monitorTask(ProceedingJoinPoint joinPoint) throws Throwable {
        long start = System.currentTimeMillis();
        try {
            return joinPoint.proceed();
        } finally {
            long duration = System.currentTimeMillis() - start;
            Metrics.timer("sync.task.duration")
                    .tag("task", joinPoint.getSignature().getName())
                    .record(duration, TimeUnit.MILLISECONDS);
        }
    }
}

@Retryable(maxAttempts = 3, backoff = @Backoff(delay = 5000))
public void performSync() {
    // 同步邏輯
}

@Recover
public void recoverSync(Exception e) {
    // 報(bào)警通知
    alertService.sendAlert("數(shù)據(jù)同步失敗：" + e.getMessage());
}

// 使用Spring Retry + RabbitMQ實(shí)現(xiàn)
@Bean
public RetryOperationsInterceptor retryInterceptor() {
    return RetryInterceptorBuilder.stateless()
            .maxAttempts(3)
            .backOffOptions(1000, 2.0, 5000)
            .recoverer(new RepublishMessageRecoverer(rabbitTemplate, "dead-letter-exchange"))
            .build();
}

# application.properties
# 定時(shí)任務(wù)配置
sync.cron=0 0 2 * * *
sync.batch-size=1000
sync.max-retry=3

# 數(shù)據(jù)源配置
spring.datasource.source.url=jdbc:mysql://source-db:3306/db
spring.datasource.source.username=user
spring.datasource.source.password=pass

spring.datasource.target.url=jdbc:mysql://target-db:3306/db
spring.datasource.target.username=user
spring.datasource.target.password=pass