Oracle to Redis 大量資料轉檔：分批並行處理與 OOM 防護實踐

全量查詢帳戶 → 全量查詢所有資料 → 批次寫入 Redis

問題：
- 10,000 帳戶 × 5 種資料 = 數十萬筆，一次塞進記憶體
- JVM Heap 峰值達 3.5GB，頻繁觸發 Full GC，偶發 OOM
- 執行時間 45 分鐘
- 失敗率約 5%（OOM 或超時導致）

批次分割層
    帳戶清單按 batchSize=50 切割
          ↓
並行控制層
    Semaphore 限制同時執行的批次數
    CompletableFuture + 自訂執行緒池
          ↓
資料處理層（單批次內）
    批次查詢（1 SQL 查 50 帳戶，避免 N+1）
    依帳戶分組
    逐帳戶寫入 Redis（錯誤隔離）
    明確釋放記憶體

public long transferByBatch(int monthsBack) throws InterruptedException {

    // 1. 查詢帳戶清單
    List<AccountEntity> accounts = accountRepo.findByDateRange(dateRange);

    // 2. 初始化控制結構
    Semaphore semaphore = new Semaphore(semaphoreNumber); // 限制並行批次數
    AtomicInteger successCount = new AtomicInteger(0);
    AtomicBoolean oomDetected = new AtomicBoolean(false);
    ErrorCollector errorCollector = new ErrorCollector(maxErrorLogCount);

    List<CompletableFuture<Boolean>> futures = new ArrayList<>();

    for (int i = 0; i < accounts.size(); i += batchSize) {

        // OOM 偵測：提前終止
        if (oomDetected.get()) break;

        List<AccountEntity> batch = accounts.subList(
            i, Math.min(i + batchSize, accounts.size())
        );

        semaphore.acquire(); // 取得許可（超過上限則阻塞等待）

        CompletableFuture<Boolean> future = transferService
            .redisTransfer(batch, successCount, oomDetected, errorCollector);

        future.whenComplete((res, ex) -> semaphore.release()); // 完成後釋放許可
        futures.add(future);

        // 每 100 批次建議 GC
        if (i % (batchSize * 100) == 0) {
            System.gc();
        }
    }

    // 等待所有批次完成
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();

    // 根據是否觸發 OOM 決定最終狀態
    if (oomDetected.get()) {
        handleOomTermination(successCount.get(), accounts.size());
    } else {
        statusService.setStatus(Status.COMPLETE);
    }

    return successCount.get();
}

@Async("accountTaskExecutor")
public CompletableFuture<Boolean> redisTransfer(
        List<AccountEntity> batch,
        AtomicInteger successCount,
        AtomicBoolean oomDetected,
        ErrorCollector errorCollector) {

    try {
        // 單批次內執行，帶帳號級錯誤隔離
        Set<String> successful = processAccountBatch(batch, errorCollector);

        successCount.addAndGet(successful.size());
        return CompletableFuture.completedFuture(true);

    } catch (Exception e) {
        // 整批失敗時回滾
        batch.forEach(this::rollbackEntity);

        if (isOomError(e)) {
            oomDetected.set(true); // 通知上層提前終止
        }

        return CompletableFuture.completedFuture(false);
    }
}

public Set<String> processAccountBatch(List<AccountEntity> accounts, ErrorCollector collector) {

    List<String> entityIdList = accounts.stream().map(AccountEntity::getEntityId).toList();
    Set<String> successful = new HashSet<>(entityIdList);
    Set<String> failed = new HashSet<>();

    // 批次查詢（1 SQL，而非 N 次）
    List<DataRecord> dataList = dataRepo.findByEntityIds(entityIdList, dataDate);

    // 依主體 ID 分組
    Map<String, List<DataRecord>> grouped = dataList.stream()
        .collect(Collectors.groupingBy(DataRecord::getEntityId));

    // 明確釋放原始清單（幫助 GC）
    dataList.clear();
    dataList = null;

    // 逐主體處理
    for (Map.Entry<String, List<DataRecord>> entry : grouped.entrySet()) {
        String entityId = entry.getKey();

        if (failed.contains(entityId)) continue; // 已在前面的階段失敗

        try {
            redisDao.save(entityId, entry.getValue());
        } catch (Exception e) {
            collector.addError(e, "data write", entityId);
            successful.remove(entityId);
            failed.add(entityId);
        } finally {
            // 每筆處理完立即清理
            entry.getValue().clear();
        }
    }

    // 回滾失敗主體
    failed.forEach(entityId -> rollbackEntityData(entityId));

    return successful;
}

@Bean(name = "accountTaskExecutor")
public Executor accountTaskExecutor() {
    ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
    executor.setCorePoolSize(corePoolSize);    // 20
    executor.setMaxPoolSize(maxPoolSize);      // 50
    executor.setQueueCapacity(queueCapacity);  // 500
    executor.setThreadNamePrefix("Acct-Trans-");

    // CallerRunsPolicy：佇列滿時由呼叫者執行 → 提供自然背壓
    executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    executor.initialize();
    return executor;
}

List<DataRecord> dataList = null;
    Map<String, List<DataRecord>> grouped = null;

    try {
        dataList = dataRepo.findByEntityIds(entityIdList, dataDate);
        grouped = dataList.stream()
            .collect(Collectors.groupingBy(DataRecord::getEntityId));

        // 原始清單用完立即釋放
        dataList.clear();
        dataList = null; // 明確置 null，讓 GC 知道可以回收

    // 繼續處理 grouped...

} finally {
    if (grouped != null) {
        grouped.clear();
        grouped = null;
    }
}

private boolean isOomError(Exception e) {
    Throwable cause = e.getCause();
    return cause instanceof RedisCommandExecutionException
        && cause.getMessage() != null
        && cause.getMessage().contains("OOM");
}

-Xms2g -Xmx4g          # 預留足夠空間
-XX:+UseG1GC            # G1 適合大 Heap + 低停頓需求
-XX:MaxGCPauseMillis=200
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=/logs/heapdump.hprof

private void executeWithRetry(Runnable operation) {
    int retryCount = 0;
    long waitMs = 500;

    while (retryCount < MAX_RETRIES) {
        try {
            operation.run();
            return;
        } catch (Exception e) {
            retryCount++;
            if (retryCount >= MAX_RETRIES) throw e;

            Thread.sleep(waitMs);
            waitMs = Math.min(waitMs * 2, 5000); // 指數退避，上限 5 秒
        }
    }
}