new ThreadPoolExecutor(
    corePoolSize,      // 核心線程數(shù)：始終存活的線程
    maximumPoolSize,   // 最大線程數(shù)：隊(duì)列滿了之后擴(kuò)展到的上限
    keepAliveTime,     // 空閑線程存活時(shí)間
    timeUnit,          // 時(shí)間單位
    workQueue,         // 任務(wù)隊(duì)列
    threadFactory,     // 線程工廠（自定義線程名稱）
    rejectedHandler    // 拒絕策略
);

提交任務(wù)
  ├── 當(dāng)前線程數(shù) < corePoolSize → 創(chuàng)建新核心線程執(zhí)行
  ├── 當(dāng)前線程數(shù) >= corePoolSize → 放入 workQueue
  ├── workQueue 已滿 且 當(dāng)前線程數(shù) < maximumPoolSize → 創(chuàng)建非核心線程執(zhí)行
  └── workQueue 已滿 且 當(dāng)前線程數(shù) >= maximumPoolSize → 執(zhí)行拒絕策略

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;

/**
 * 異步導(dǎo)入線程池配置.
 */
@Configuration
public class AsyncImportThreadPoolConfig {

  @Bean(name = "importThreadPool", destroyMethod = "shutdown")
  public ThreadPoolExecutor importThreadPool() {
    return new ThreadPoolExecutor(
        4,                              // 核心線程數(shù)
        8,                              // 最大線程數(shù)
        60, TimeUnit.SECONDS,           // 空閑線程存活60秒
        new ArrayBlockingQueue<>(16),   // 有界隊(duì)列，最多堆積16個(gè)任務(wù)
        new ImportThreadFactory(),      // 自定義線程工廠
        new ThreadPoolExecutor.CallerRunsPolicy()  // 隊(duì)列滿時(shí)由調(diào)用線程執(zhí)行
    );
  }

  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); // 非守護(hù)線程，確保任務(wù)執(zhí)行完
      return t;
    }
  }
}

package com.example.service;

import java.util.List;

/**
 * 批量導(dǎo)入服務(wù)接口.
 */
public interface BatchImportService {

  /**
   * 導(dǎo)入數(shù)據(jù)：同步校驗(yàn) + 異步入庫(kù).
   *
   * @param rawDataList 原始數(shù)據(jù)列表（已從文件中解析出來(lái)）
   * @param operatorId  操作人ID
   * @return 導(dǎo)入結(jié)果提示
   */
  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) {

    // ========== 第一步：同步校驗(yàn)（在請(qǐng)求線程中執(zhí)行） ==========
    for (int i = 0; i < rawDataList.size(); i++) {
      RawData raw = rawDataList.get(i);
      String error = validate(raw);
      if (error != null) {
        // 遇到第一條錯(cuò)誤立即中斷，同步返回給前端
        throw new RuntimeException("第" + (i + 1) + "行：" + error);
      }
    }

    // ========== 第二步：數(shù)據(jù)轉(zhuǎn)換 ==========
    List<ImportRecord> recordList = new ArrayList<>(rawDataList.size());
    for (RawData raw : rawDataList) {
      ImportRecord record = convertToEntity(raw, operatorId);
      recordList.add(record);
    }

    // ========== 第三步：異步批量插入（提交到線程池） ==========
    // 注意：這里的 recordList 對(duì)象引用傳遞給了異步線程
    // 確保主線程之后不再修改這個(gè)列表
    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("異步導(dǎo)入完成，共{}條，耗時(shí){}ms", total, cost);
      } catch (Exception e) {
        log.error("異步導(dǎo)入失敗", e);
        // 可選：更新主表狀態(tài)為"導(dǎo)入失敗"
      }
    });

    // ========== 第四步：同步返回成功提示 ==========
    return "導(dǎo)入任務(wù)已提交，共" + recordList.size() + "條數(shù)據(jù)正在后臺(tái)處理";
  }

  /**
   * 校驗(yàn)單條數(shù)據(jù).
   * 返回 null 表示通過(guò)，返回錯(cuò)誤信息表示失敗.
   */
  private String validate(RawData raw) {
    if (raw.getAmount() == null) {
      return "數(shù)量不能為空";
    }
    if (raw.getAmount() <= 0 || raw.getAmount() > 999999) {
      return "數(shù)量必須為大于0的正整數(shù)，最多六位";
    }
    return null;
  }

  /**
   * 原始數(shù)據(jù)轉(zhuǎn)換為實(shí)體.
   */
  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;
  }
}

請(qǐng)求線程                          線程池工作線程
   │                                  │
   │── 解析文件 ──→                    │
   │── 逐行校驗(yàn) ──→                    │
   │   (校驗(yàn)不過(guò)直接返回錯(cuò)誤)             │
   │── 轉(zhuǎn)換數(shù)據(jù) ──→                    │
   │── threadPool.execute(task) ──→   │
   │                                  │── 批量INSERT第1批(2000條)
   │← 返回"導(dǎo)入任務(wù)已提交" ──           │── 批量INSERT第2批(2000條)
   │                                  │── ...
   │   (HTTP響應(yīng)已返回給前端)            │── 批量INSERT第N批
   │                                  │── 記錄日志"導(dǎo)入完成"

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

log.info("線程池狀態(tài) - 活躍:{}, 隊(duì)列積壓:{}, 已完成:{}",
    threadPool.getActiveCount(),
    threadPool.getQueue().size(),
    threadPool.getCompletedTaskCount());