Spring Batch迁移1亿条数据时HTTP连接超时问题咨询

业务场景与问题描述 #

• 需求：从共享对象存储桶加载1亿条记录至MongoDB
• 当前实现：基于HTTP Range 请求头做分区处理，使用15个线程并行加载，总耗时约30分钟
• 核心问题：网络设备会在15分钟后自动关闭HTTP连接，导致加载中断

当前使用的连接资源代码如下：

HttpURLConnection httpConnection = null;
try {
    httpConnection = (HttpURLConnection) this.url.openConnection();
    ResourceUtils.useCachesIfNecessary(httpConnection);
    if(StringUtils.hasText(byteRangeHeader)) {
        httpConnection.setRequestProperty("Range", String.format("bytes=%s", byteRangeHeader));
    }
    inputStream = httpConnection.getInputStream();

} catch (Exception e) {
    e.printStacktrace
}
return inputStream;

用户疑问：

1. 是否可以先将1亿条记录加载至内存后再处理（计划采用多进程+多线程方式）
2. 能否在15分钟前刷新连接，且新连接建立后Spring Batch读取器能从上次完成的位置继续？

解决方案与疑问解答 #

关于疑问1：全量加载至内存的方案 #

不推荐采用该方案，原因如下：

• 内存压力过大：1亿条记录即使每条仅占1KB，总容量也达100GB，远超常规服务器内存上限，必然触发OOM（内存溢出）
• 可靠性差：若加载过程中出现进程崩溃、机器故障，已加载的数据全部丢失，需重新从头开始
• 多进程/多线程无法解决内存瓶颈，反而会增加进程间通信的额外开销，降低整体效率

关于疑问2：Spring Batch断点续读+连接刷新方案 #

Spring Batch完全支持断点续读，但需要结合你的Range请求逻辑做适配，具体实现步骤：

1. 记录已处理字节偏移量：在分区处理过程中，实时将当前已读取并处理完成的字节位置存入Spring Batch的ExecutionContext（或外部存储如Redis、数据库）
2. 主动控制连接存活时间：设置连接超时和读取超时时间小于15分钟（例如14分钟），当到达超时阈值时，主动关闭当前连接
3. 基于偏移量重建连接：新建连接时，将Range请求头设置为bytes=已处理偏移量-分区结束偏移量，Spring Batch读取器从新的输入流继续读取数据
4. 启用Spring Batch重启机制：确保Job的元数据（如StepExecution）正确持久化，即使连接中断，重启Job后可直接从上次记录的偏移量继续处理

额外优化方案 #

1. 缩小分区粒度

将原有分区拆分为更小的单元，确保每个分区的处理时间控制在10分钟以内，从根源上避免单个连接存活时间超过15分钟。例如：若原分区每个处理30分钟，可拆分为3个小分区，每个处理10分钟。

2. 使用对象存储SDK替代原生HttpURLConnection

多数云厂商的对象存储SDK（如AWS S3、阿里云OSS）内置了连接池、自动重试、断点续传功能，能自动处理连接超时和重连逻辑，无需手动维护Range请求和连接状态，大幅降低开发复杂度。

3. 增加超时重试机制

在代码中捕获连接超时异常，自动基于记录的偏移量重新发起Range请求，同时设置重试次数上限，避免无限重试。

优化后的示例代码 #

HttpURLConnection httpConnection = null;
InputStream inputStream = null;
// 从Spring Batch的ExecutionContext获取已处理偏移量，默认0
Long processedOffset = executionContext.getLong("processedOffset", 0L);
// 假设当前分区的结束偏移量已预先定义
Long partitionEndOffset = getPartitionEndOffset();

try {
    httpConnection = (HttpURLConnection) this.url.openConnection();
    ResourceUtils.useCachesIfNecessary(httpConnection);
    // 设置连接和读取超时为14分钟（840000毫秒），小于15分钟的设备超时时间
    httpConnection.setConnectTimeout(840000);
    httpConnection.setReadTimeout(840000);
    
    // 构造Range请求头，从已处理位置开始
    String rangeHeader = String.format("bytes=%d-%d", processedOffset, partitionEndOffset);
    httpConnection.setRequestProperty("Range", rangeHeader);
    inputStream = httpConnection.getInputStream();
    
    // 读取并处理数据，实时更新已处理偏移量
    byte[] buffer = new byte[4096];
    int bytesRead;
    while ((bytesRead = inputStream.read(buffer)) != -1) {
        // 处理读取到的记录逻辑
        processRecords(buffer, bytesRead);
        // 更新偏移量并写入ExecutionContext
        processedOffset += bytesRead;
        executionContext.putLong("processedOffset", processedOffset);
    }

} catch (SocketTimeoutException e) {
    // 抛出可重试异常，Spring Batch会根据配置重启该Step
    throw new RetryableException("Connection timeout, will retry from last position", e);
} catch (Exception e) {
    e.printStackTrace();
    throw new RuntimeException("Failed to load data from object storage", e);
} finally {
    // 关闭资源
    if (inputStream != null) {
        try {
            inputStream.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
    if (httpConnection != null) {
        httpConnection.disconnect();
    }
}

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