Java分批处理实战：基于subList实现数据分片入库

// JDK8+，异常处理与性能注释
public void batchInsert(List<Data> dataList, int batchSize) {
    if (dataList == null || dataList.isEmpty() || batchSize <= 0) {
        throw new IllegalArgumentException("参数非法");
    }
    int total = dataList.size();
    int batchCount = (total + batchSize - 1) / batchSize; // 分片数量
    for (int i = 0; i < batchCount; i++) {
        int fromIndex = i * batchSize;
        int toIndex = Math.min(fromIndex + batchSize, total);
        List<Data> batchList = dataList.subList(fromIndex, toIndex);
        try {
            // 数据库批量入库操作
            saveBatch(batchList);
        } catch (Exception e) {
            // 监控与补偿机制（如重试、告警）
            log.error("批次{}入库失败，from:{} to:{}", i, fromIndex, toIndex, e);
            // 可选：失败批次记录到补偿队列
        }
    }
}

split(data, 500).parallelStream()
    .forEach(batch -> repository.bulkInsert(batch));

// 动态调整分片大小
// 动态分片算法（根据系统负载调整）
public static <T> List<List<T>> adaptiveSplit(List<T> list, int baseSize) {
    int currentLoad = getSystemLoad(); // 获取当前系统负载
    int adjustedSize = currentLoad > 70 ? baseSize / 2 : baseSize;
    return split(list, adjustedSize);
}

// 带事务补偿的分批处理
public void batchInsertWithRetry(List<Data> data) {
    List<List<Data>> batches = split(data, 1000);
    
    batches.forEach(batch -> {
        int retryCount = 0;
        while (retryCount < 3) {
            try {
                transactionalTemplate.execute(status -> {
                    repository.batchInsert(batch);
                    return Boolean.TRUE;
                });
                break;
            } catch (DataAccessException e) {
                retryCount++;
                log.warn("批次插入失败，重试次数: {}", retryCount);
            }
        }
    });
}

// 视图特性示例
List<Integer> original = Arrays.asList(1, 2, 3, 4, 5);
List<Integer> sub = original.subList(1, 3); // [2, 3]
sub.set(0, 99); // 修改视图
System.out.println(original); // [1, 99, 3, 4, 5] - 原列表也被修改！

int batchSize = Math.min(
    1000,                              // 数据库推荐值
    Runtime.getRuntime().maxMemory() / 1024 / 100  // 占用内存 < 1MB
);

// 按批次顺序执行，保证数据顺序
for (int i = 0; i < batchCount; i++) {
    processBatch(batches.get(i));  // 串行执行
}

// 添加批次序号，最终按序号排序
AtomicInteger batchIndex = new AtomicInteger(0);
batches.parallelStream()
    .forEach(batch -> {
        int index = batchIndex.getAndIncrement();
        batch.setIndex(index);
        processBatch(batch);
    });

// 完整的异常处理示例
public void safeBatchInsert(List<Data> data) {
    List<List<Data>> batches = split(data, 1000);

    for (int i = 0; i < batches.size(); i++) {
        List<Data> batch = batches.get(i);
        int retryCount = 0;
        boolean success = false;

        while (retryCount < 3 && !success) {
            try {
                // 指数退避：第一次立即重试，后续等待
                if (retryCount > 0) {
                    Thread.sleep(1000L * (1L << retryCount));
                }

                saveBatch(batch);
                success = true;

            } catch (SQLException e) {
                retryCount++;
                log.warn("批次 {} 第 {} 次重试", i, retryCount);

                // 最后一次重试失败，记录到补偿队列
                if (retryCount >= 3) {
                    retryQueue.add(new RetryTask(i, batch, e));
                }
            }
        }
    }
}