首页 > 软件编程 > java > Java大数据处理

Java中大数据量批处理的性能优化指南

2026-07-07 08:50:08 作者：霸道流氓气质

批处理的核心原则是减少网络和SQL解析开销,本文总结了Java大数据量批处理性能优化的核心方法,重点对比了不同方案的效率差异,文中的示例代码讲解详细,感兴趣的小伙伴可以了解下

一、批处理核心原则

单条处理：20万次网络往返 + 20万次SQL解析 = 极慢
批量处理：200次网络往返 + 200次SQL解析 = 快1000倍

批次大小选择平衡点：

太小（1~100）：网络往返次数多，总耗时长
太大（10000+）：单次内存占用高，SQL 过长，数据库锁持有时间长
最佳区间：1000~5000

二、导入优化

2.1 批量 INSERT（核心）

// 慢：逐条 INSERT（20万次网络IO）
for (Entity e : list) {
    mapper.insert(e);  // 每条一次网络往返
}

// 快：批量 INSERT（100次网络IO）
for (int i = 0; i < list.size(); i += 2000) {
    int end = Math.min(i + 2000, list.size());
    mapper.batchInsert(list.subList(i, end));
}

XML：

<insert id="batchInsert">
  INSERT INTO my_table (col1, col2, col3) VALUES
  <foreach collection="list" item="item" separator=",">
    (#{item.col1}, #{item.col2}, #{item.col3})
  </foreach>
</insert>

2.2 分页处理模式

int pageSize = 2000;
int page = 1;
while (true) {
    List<Detail> batch = mapper.selectPage(taskId, page, pageSize);
    if (batch.isEmpty()) break;

    // 校验或入库
    processBatch(batch);

    // 回写状态
    mapper.batchUpdateStatus(batch);
    page++;
}

2.3 完整导入示例

public void importData(List<RawData> allData) {
    int batchSize = 2000;
    int total = allData.size();

    for (int i = 0; i < total; i += batchSize) {
        int end = Math.min(i + batchSize, total);
        List<RawData> batch = allData.subList(i, end);

        // 1. 校验
        List<Entity> validList = new ArrayList<>();
        for (RawData raw : batch) {
            if (validate(raw)) {
                validList.add(convert(raw));
            }
        }

        // 2. 批量入库
        if (!validList.isEmpty()) {
            mapper.batchInsert(validList);
        }
    }
}

三、导出优化

3.1 SXSSFWorkbook 流式写入

SXSSFWorkbook workbook = new SXSSFWorkbook(200); // 窗口200行
Sheet sheet = workbook.createSheet("数据");

int pageSize = 5000;
int page = 1;
int rowIndex = 1;

while (true) {
    List<Entity> batch = mapper.selectPage(page, pageSize);
    if (batch.isEmpty()) break;

    for (Entity e : batch) {
        Row row = sheet.createRow(rowIndex++);
        row.createCell(0).setCellValue(e.getName());
        row.createCell(1).setCellValue(e.getAmount());
    }
    page++;
}

workbook.write(outputStream);
workbook.dispose(); // 清理临时文件

3.2 EasyExcel 流式写入

ExcelWriter writer = EasyExcelFactory.write(outputStream, ExportRow.class).build();
WriteSheet sheet = EasyExcelFactory.writerSheet("数据").build();

int page = 0;
while (true) {
    List<ExportRow> batch = queryData(param, page, 5000);
    if (batch.isEmpty()) break;
    writer.write(batch, sheet);
    page++;
}
writer.finish();

四、方案对比

导入方案

方案	20万条耗时	内存	复杂度
逐条INSERT	15~30分钟	低	低
批量INSERT(2000/批)	5~10秒	中(25MB)	低
MQ异步+批量INSERT	5~10秒(用户无感)	中	高
线程池异步+批量INSERT	3~8秒(用户无感)	中	中

导出方案

方案	20万条耗时	内存	文件大小
XSSFWorkbook全内存	15~25秒	400MB+	8~12MB
SXSSFWorkbook+分页查	4~8秒	20~30MB	8~12MB
EasyExcel流式	3~6秒	15~25MB	8~12MB
CSV	1~3秒	<5MB	15~20MB

五、完整示例

@Service
public class BatchProcessService {

  @Resource
  private DataMapper dataMapper;

  private static final int BATCH_SIZE = 2000;

  /** 批量导入. */
  public void batchImport(List<ImportRow> rawData) {
    List<DataEntity> buffer = new ArrayList<>(BATCH_SIZE);

    for (ImportRow raw : rawData) {
      if (!validate(raw)) continue;
      buffer.add(convert(raw));

      if (buffer.size() >= BATCH_SIZE) {
        dataMapper.batchInsert(buffer);
        buffer.clear();
      }
    }
    // 处理剩余
    if (!buffer.isEmpty()) {
      dataMapper.batchInsert(buffer);
    }
  }

  /** 流式导出. */
  public void streamExport(OutputStream out, QueryParam param) {
    SXSSFWorkbook wb = new SXSSFWorkbook(200);
    try {
      Sheet sheet = wb.createSheet("导出数据");
      writeHeader(sheet);

      int page = 1, rowIdx = 1;
      while (true) {
        List<DataEntity> batch = dataMapper.selectByPage(param, page, 5000);
        if (batch.isEmpty()) break;
        for (DataEntity e : batch) {
          writeRow(sheet, rowIdx++, e);
        }
        page++;
      }
      wb.write(out);
    } finally {
      wb.dispose();
    }
  }
}

六、关键配置

# 数据库连接池（批处理需要足够连接）
spring:
  datasource:
    hikari:
      maximum-pool-size: 20      # 并发批处理时需要足够连接
      connection-timeout: 30000
# MySQL批量操作需要的URL参数
      url: jdbc:mysql://host:3306/db?rewriteBatchedStatements=true
      #                               ↑ 关键！开启批量重写，性能提升5~10倍

rewriteBatchedStatements=true 让 JDBC 驱动将多条 INSERT 合并为一条多值 INSERT，是批量写入性能的关键配置。

七、批量 UPDATE 优化

7.1 逐条 vs 批量

// 慢：逐条UPDATE
for (Detail d : list) {
    mapper.updateStatus(d.getId(), d.getStatus());  // 20万次
}

// 快：CASE WHEN 批量UPDATE
mapper.batchUpdateStatus(list);  // 按批次执行

XML 实现：

<update id="batchUpdateStatus">
  UPDATE exchange_task_detail
  SET status = CASE id
  <foreach collection="list" item="item">
    WHEN #{item.id} THEN #{item.status}
  </foreach>
  END,
  data = CASE id
  <foreach collection="list" item="item">
    WHEN #{item.id} THEN #{item.data}
  </foreach>
  END
  WHERE id IN
  <foreach collection="list" item="item" open="(" separator="," close=")">
    #{item.id}
  </foreach>
</update>

7.2 另一种方式：临时表 JOIN UPDATE

数据量极大时（50万+），可以先把要更新的数据写入临时表，再用 JOIN UPDATE：

-- 1. 创建临时表并插入数据
CREATE TEMPORARY TABLE tmp_update (id INT, status INT);
INSERT INTO tmp_update VALUES (1, 3), (2, 3), ...;
-- 2. JOIN UPDATE
UPDATE exchange_task_detail d
JOIN tmp_update t ON d.id = t.id
SET d.status = t.status;

八、并发批处理

8.1 线程池分片

public void parallelImport(List<Entity> allData) {
    int batchSize = 2000;
    ExecutorService executor = Executors.newFixedThreadPool(4);
    List<Future<?>> futures = new ArrayList<>();

    for (int i = 0; i < allData.size(); i += batchSize) {
        int start = i;
        int end = Math.min(i + batchSize, allData.size());
        List<Entity> batch = allData.subList(start, end);

        futures.add(executor.submit(() -> {
            mapper.batchInsert(batch);
        }));
    }

    // 等待全部完成
    for (Future<?> f : futures) {
        f.get();
    }
    executor.shutdown();
}

注意：并发写入时需确保没有唯一键冲突，且数据库连接池大小 >= 线程数。

8.2 性能对比

方式	20万条INSERT耗时	数据库连接占用
单线程 2000/批	5~10秒	1个
4线程 2000/批	2~4秒	4个
8线程 2000/批	1~2.5秒	8个

九、内存优化技巧

9.1 避免全量加载

// 错误：20万条全加载到内存
List<Entity> all = mapper.selectAll();  // ~200MB

// 正确：分页加载，用完即弃
int page = 1;
while (true) {
    List<Entity> batch = mapper.selectPage(page, 5000);  // ~5MB
    if (batch.isEmpty()) break;
    process(batch);
    page++;
    // batch 在下一次循环时可被GC
}

9.2 流式读取 Excel（导入时）

// EasyExcel 流式读取，不会把20万行全加载到内存
EasyExcelFactory.read(inputStream, RowData.class,
    new PageReadListener<RowData>(batch -> {
        // 每100行回调一次，处理后释放
        processBatch(batch);
    }, 100))
    .sheet()
    .doRead();

十、rewriteBatchedStatements详解

这是 MySQL JDBC 驱动的关键参数：jdbc:mysql://host:3306/db?rewriteBatchedStatements=true

设置	实际执行的SQL	网络往返
false（默认）	`INSERT INTO t VALUES(1); INSERT INTO t VALUES(2); ...`	N次
true	`INSERT INTO t VALUES(1),(2),(3),...`	1次

对批量 INSERT 性能提升 5~10倍。对 UPDATE 也有类似优化（合并为多条一次发送）。

十一、总结

导入优化公式：

总耗时 = (数据量 / 批次大小) × 单批耗时
单批耗时 = 网络往返(~2ms) + SQL执行(~5ms) + 业务处理(~1ms)
20万条 / 2000批 × 8ms = 0.8秒（理想值）
实际加上GC、连接等待等 ≈ 5~10秒

导出优化公式：

总耗时 = 查询时间 + 写入时间
查询 = (数据量 / 页大小) × 单页查询时间
写入 = SXSSFWorkbook 流式写（不受内存限制）

批次大小选择：

┌─────────────┬──────────────────────────────┐
│ 数据量 │ 推荐批次大小 │
├─────────────┼──────────────────────────────┤
│ < 1万 │ 1000 │
│ 1万~10万 │ 2000 │
│ 10万~50万 │ 2000~5000 │
│ > 50万 │ 5000 + 并发线程池 │
└─────────────┴──────────────────────────────┘

以上就是Java中大数据量批处理的性能优化指南的详细内容，更多关于Java大数据处理的资料请关注脚本之家其它相关文章！