Java大数据量异步处理方案

new ThreadPoolExecutor(
    corePoolSize,      // 核心线程数：始终存活的线程
    maximumPoolSize,   // 最大线程数：队列满了之后扩展到的上限
    keepAliveTime,     // 空闲线程存活时间
    timeUnit,          // 时间单位
    workQueue,         // 任务队列
    threadFactory,     // 线程工厂（自定义线程名称）
    rejectedHandler    // 拒绝策略
);

提交任务
  ├── 当前线程数 < corePoolSize → 创建新核心线程执行
  ├── 当前线程数 >= corePoolSize → 放入 workQueue
  ├── workQueue 已满 且 当前线程数 < maximumPoolSize → 创建非核心线程执行
  └── workQueue 已满 且 当前线程数 >= maximumPoolSize → 执行拒绝策略

package com.example.config;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

/**
 * 异步导入线程池配置.
 */
@Configuration
public class AsyncImportThreadPoolConfig {

  @Bean(name = "importThreadPool", destroyMethod = "shutdown")
  public ThreadPoolExecutor importThreadPool() {
    return new ThreadPoolExecutor(
        4,                              // 核心线程数
        8,                              // 最大线程数
        60, TimeUnit.SECONDS,           // 空闲线程存活60秒
        new ArrayBlockingQueue<>(16),   // 有界队列，最多堆积16个任务
        new ImportThreadFactory(),      // 自定义线程工厂
        new ThreadPoolExecutor.CallerRunsPolicy()  // 队列满时由调用线程执行
    );
  }

  static class ImportThreadFactory implements ThreadFactory {
    private final AtomicInteger counter = new AtomicInteger(1);

    @Override
    public Thread newThread(Runnable r) {
      Thread t = new Thread(r, "import-worker-" + counter.getAndIncrement());
      t.setDaemon(false); // 非守护线程，确保任务执行完
      return t;
    }
  }
}

package com.example.service;

import java.util.List;

/**
 * 批量导入服务接口.
 */
public interface BatchImportService {

  /**
   * 导入数据：同步校验 + 异步入库.
   *
   * @param rawDataList 原始数据列表（已从文件中解析出来）
   * @param operatorId  操作人ID
   * @return 导入结果提示
   */
  String importData(List<RawData> rawDataList, String operatorId);
}

package com.example.service.impl;

import com.example.entity.ImportRecord;
import com.example.mapper.ImportRecordMapper;
import com.example.service.BatchImportService;
import jakarta.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Service;

@Slf4j
@Service
public class BatchImportServiceImpl implements BatchImportService {

  private static final int BATCH_SIZE = 2000;

  @Resource
  @Qualifier("importThreadPool")
  private ThreadPoolExecutor threadPool;

  @Resource
  private ImportRecordMapper importRecordMapper;

  @Override
  public String importData(List<RawData> rawDataList, String operatorId) {

    // ========== 第一步：同步校验（在请求线程中执行） ==========
    for (int i = 0; i < rawDataList.size(); i++) {
      RawData raw = rawDataList.get(i);
      String error = validate(raw);
      if (error != null) {
        // 遇到第一条错误立即中断，同步返回给前端
        throw new RuntimeException("第" + (i + 1) + "行：" + error);
      }
    }

    // ========== 第二步：数据转换 ==========
    List<ImportRecord> recordList = new ArrayList<>(rawDataList.size());
    for (RawData raw : rawDataList) {
      ImportRecord record = convertToEntity(raw, operatorId);
      recordList.add(record);
    }

    // ========== 第三步：异步批量插入（提交到线程池） ==========
    // 注意：这里的 recordList 对象引用传递给了异步线程
    // 确保主线程之后不再修改这个列表
    threadPool.execute(() -> {
      try {
        long start = System.currentTimeMillis();
        int total = recordList.size();

        for (int i = 0; i < total; i += BATCH_SIZE) {
          int end = Math.min(i + BATCH_SIZE, total);
          List<ImportRecord> batch = recordList.subList(i, end);
          importRecordMapper.batchInsert(batch);
        }

        long cost = System.currentTimeMillis() - start;
        log.info("异步导入完成，共{}条，耗时{}ms", total, cost);
      } catch (Exception e) {
        log.error("异步导入失败", e);
        // 可选：更新主表状态为"导入失败"
      }
    });

    // ========== 第四步：同步返回成功提示 ==========
    return "导入任务已提交，共" + recordList.size() + "条数据正在后台处理";
  }

  /**
   * 校验单条数据.
   * 返回 null 表示通过，返回错误信息表示失败.
   */
  private String validate(RawData raw) {
    if (raw.getAmount() == null) {
      return "数量不能为空";
    }
    if (raw.getAmount() <= 0 || raw.getAmount() > 999999) {
      return "数量必须为大于0的正整数，最多六位";
    }
    return null;
  }

  /**
   * 原始数据转换为实体.
   */
  private ImportRecord convertToEntity(RawData raw, String operatorId) {
    ImportRecord record = new ImportRecord();
    record.setCode(raw.getCode());
    record.setName(raw.getName());
    record.setAmount(raw.getAmount());
    record.setOperatorId(operatorId);
    return record;
  }
}

请求线程                          线程池工作线程
   │                                  │
   │── 解析文件 ──→                    │
   │── 逐行校验 ──→                    │
   │   (校验不过直接返回错误)             │
   │── 转换数据 ──→                    │
   │── threadPool.execute(task) ──→   │
   │                                  │── 批量INSERT第1批(2000条)
   │← 返回"导入任务已提交" ──           │── 批量INSERT第2批(2000条)
   │                                  │── ...
   │   (HTTP响应已返回给前端)            │── 批量INSERT第N批
   │                                  │── 记录日志"导入完成"

@PreDestroy
public void destroy() {
    threadPool.shutdown();
    try {
        if (!threadPool.awaitTermination(60, TimeUnit.SECONDS)) {
            threadPool.shutdownNow();
        }
    } catch (InterruptedException e) {
        threadPool.shutdownNow();
    }
}

log.info("线程池状态 - 活跃:{}, 队列积压:{}, 已完成:{}",
    threadPool.getActiveCount(),
    threadPool.getQueue().size(),
    threadPool.getCompletedTaskCount());