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

// 慢：逐条 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));
}

<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>

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++;
}

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);
        }
    }
}

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(); // 清理临时文件

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();

@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倍

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

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

<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>

-- 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;

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();
}

// 错误：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
}

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