EasyExcel SAX 流式读取 + 有界线程池 + CallerRunsPolicy——一个 Java 后端实习生的“导入三件套“实战

weixin_52201538

252人浏览 · 2026-07-14 10:17:05

weixin_52201538 · 2026-07-14 10:17:05 发布

EasyExcel SAX 流式读取 + 有界线程池 + CallerRunsPolicy——一个 Java 后端实习生的"导入三件套"实战

副标题：10w 行 Excel 不爆内存、不丢任务、自动降速：组合拳就这么打

一、开场引入：为什么"Excel 导入"对一个后端实习生是噩梦？

实习期间，业务让我做一个"线索批量导入"功能——用户上传 5w 行 Excel，导入到系统里。

我第一版用最简单的 EasyExcel.read().head().sheet().doReadSync() —— OOM 了。

同事看了一眼：

“你这是一次性把整张表读到内存，5w 行 × 200 字段 = 千万级对象，JVM 顶不住。”

我一脸懵：“那咋办？”

mentor 抛下一句话：

“Excel 导入三大难题：内存爆掉、并发失控、失败回滚。靠单个组件都解决不了，必须三件套组合——SAX 流式读取 + 有界线程池 + CallerRunsPolicy。”

回去翻了项目代码，发现：

业务代码几乎不用 EasyExcel 链式 API，而是基于 POI 的 XSSFSheetXMLHandler 自封装 SAX 流式读取
导入校验环节用 ThreadFactoryBuilder 命名 + ArrayBlockingQueue(100) 有界 + CallerRunsPolicy 三件套
二十几个 Listener 子类都按这套模板来

这次我学到：EasyExcel 链式 API 不够用，必须配合自封 SAX + 线程池三件套。

今天这篇博客，我就把项目里的 “导入三件套” 掰开揉碎讲透，让更多像我一样的后端新人少走弯路。

💡 金句：Excel 导入不是"调个 API 就行"，是"组合拳工程"。

二、概念扫盲：Excel 导入的三大难题

2.1 三大难题

2.2 为什么单个组件解决不了

组件	单独使用的问题
`EasyExcel.read()` 链式	一次性读全表，5w 行 OOM
普通线程池（无界队列）	任务堆积到 100w，OOM
默认 `AbortPolicy`	队列满就抛异常，任务丢失 / 接口报错

所以——EasyExcel 不够，线程池不够，CallerRunsPolicy 不够——三个组合起来才够。

2.3 三件套组合思路

💡 金句：SAX 解决"读"，线程池解决"算"，CallerRunsPolicy 解决"稳"——三件套各司其职。

2.4 整体架构图

三、精讲 1：EasyExcel SAX 流式读取 ⭐ 重头

3.1 一句话定义

SAX（Simple API for XML）= 逐行扫描 Excel XML 流，不一次性加载到内存。

3.2 链式 API vs SAX 流式

3.3 SAX 原理（一句话）

概念	解释
DOM	一次性把 XML 全加载到内存，构建对象树
SAX	边读边触发回调（`invoke` / `invokeHead`），不构建对象树

Excel 本质是 XML 文件，POI 的 XSSFSheetXMLHandler 就是 SAX 实现。

3.4 项目里的真 SAX 抽象（脱敏）

// ✅ 项目自封的 SAX 抽象基类（简化版）
public abstract class AbstractImportExcelSaxReadParse<T> {

    // 核心：走 POI SAX，不读全内存
    public List<T> importExcel(File file, String templatePath, String json) throws Exception {
        InputStream in = Files.newInputStream(file.toPath());
        // 底层: XSSFReader + SAXHelper.newXMLReader()
        // 不会把整张 sheet 加载到内存
        return parserSheet(in, "8", templatePath, null, json);
    }

    // 子类实现：每行校验逻辑
    protected abstract void verify(T data, StringBuilder errorMsg);

    // 子类实现：批量跨表校验
    protected abstract void validateImportList(List<T> list, Map<Integer, StringBuilder> errorMap);
}

优点：

✅ 十几万行 Excel 也不爆内存
✅ 解析过程可控（每行回调）

3.5 三层 Listener 抽象（项目新派写法）

三段式处理流程：

3.6 错误行就地回写（项目亮点）

// 解析时收集错误
protected Map<Integer, StringBuilder> errorMap = new LinkedHashMap<>();

// 解析完写"标黄版 Excel"
public String uploadFile(OssHelper oss, File file, String userCode) {
    // 用 SXSSF 在错误行追加一列"错误信息"
    // 错误单元格变黄（背景色）
    // 上传到 OSS，返回 fileId
}

好处：用户下载"标黄版 Excel"自查，比文本提示友好太多。

3.7 关键点

✅ 真 SAX 不爆内存（不是 EasyExcel 链式 API）
✅ 三层 Listener 抽象：Fixed / Change / Dynamic 覆盖 90% 业务
✅ 错误就地回写 → 友好用户提示
❌ 不要在 Listener 里写重业务逻辑（应该只做基础校验）
✅ 重业务（跨表 / 权限）拆到专门的校验线程池

💡 金句：SAX 解决"读"，线程池解决"算"，CallerRunsPolicy 解决"稳"——三件套各司其职。

四、精讲 2：有界线程池 ⭐ 重头

4.1 一句话定义

有界线程池 = 任务队列容量有限的线程池，避免任务堆积导致 OOM。

4.2 为什么必须有界？

队列类型	容量	风险
`LinkedBlockingQueue()`	无界（Integer.MAX_VALUE）	⚠️ 高风险 OOM
`LinkedBlockingQueue(1000)`	有界	✅ 推荐
`ArrayBlockingQueue(100)`	严格有界	✅✅ 适合导入校验

4.3 项目里的"导入校验线程池"模板（脱敏）

// ✅ 项目里内置的有界线程池（最贴博客主题的代码）
private static final int  BATCH_SIZE       = 200;   // 每批 200 行
private static final int  MAX_THREADS      = 10;    // 最多 10 个线程
private static final int  QUEUE_SIZE       = 100;   // 有界队列
private static final long KEEP_ALIVE_TIME  = 0L;    // 固定池

private ThreadPoolExecutor createBoundedThreadPool(int corePoolSize) {
    return new ThreadPoolExecutor(
        corePoolSize, corePoolSize,                  // ★ 固定大小线程池
        KEEP_ALIVE_TIME, TimeUnit.MILLISECONDS,
        new ArrayBlockingQueue<>(QUEUE_SIZE),       // ★ 严格有界队列（防 OOM 关键）
        new ThreadFactoryBuilder()
            .setNameFormat("xxx-validate-pool-%d")    // ★ 命名线程，方便排查
            .setDaemon(true)                          // 守护线程
            .setUncaughtExceptionHandler((t, e) ->
                log.error("线程 {} 发生未捕获异常", t.getName(), e))
            .build(),
        new ThreadPoolExecutor.CallerRunsPolicy()    // ★ 背压到提交线程
    );
}

每个参数都有讲究：

参数	值	为什么这么选
`corePoolSize == maximumPoolSize`	10	固定大小，导入校验 CPU 不重，不需要弹性扩容
`ArrayBlockingQueue(100)`	100	严格有界，防 OOM；100 比 LinkedBlockingQueue 更安全
`keepAliveTime = 0`	0	固定池不需要回收，简单可控
`ThreadFactoryBuilder` 命名	`xxx-validate-pool-%d`	线程名一眼看出业务，日志排查快 10 倍
`daemon = true`	true	守护线程，JVM 退出时不阻塞
`CallerRunsPolicy`	-	背压到提交线程（见精讲 3）

4.4 项目里的"工具类工厂"（团队约定）

// ✅ 项目里所有 Spring Bean 线程池都走这个工厂
public static ThreadPoolExecutor createThreadPool(
        int corePoolSize, int maximumPoolSize, int queueCapacity,
        long keepAliveTime, String namePrefix) {

    return new ThreadPoolExecutor(
        corePoolSize, maximumPoolSize, keepAliveTime, TimeUnit.SECONDS,
        new LinkedBlockingQueue<>(queueCapacity),    // 有界
        new ThreadFactory() {
            private final AtomicInteger t = new AtomicInteger(1);
            @Override public Thread newThread(Runnable r) {
                Thread th = new Thread(r);
                th.setName(namePrefix + "-" + t.getAndIncrement());
                return th;
            }
        },
        new ThreadPoolExecutor.CallerRunsPolicy()    // 团队默认
    );
}

4.5 用 CompletableFuture 提交任务（实战）

// 数据量小 → 单线程，避免线程切换
if (totalSize <= BATCH_SIZE) {
    validateBatch(list, list, 0, errorMap, ...);
    return;
}

// 数据量大 → 计算最优线程数
int optimalThreads = Math.min((totalSize + BATCH_SIZE - 1) / BATCH_SIZE, MAX_THREADS);
ThreadPoolExecutor executor = createBoundedThreadPool(optimalThreads);
AtomicInteger completedBatches = new AtomicInteger(0);
int totalBatches = (totalSize + BATCH_SIZE - 1) / BATCH_SIZE;

try {
    List<CompletableFuture<Void>> futures = new ArrayList<>();

    for (int i = 0; i < totalSize; i += BATCH_SIZE) {
        final int start = i;
        final List<DTO> batch = list.subList(start, Math.min(i + BATCH_SIZE, totalSize));

        CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
            validateBatch(list, batch, start, errorMap, ...);
            completedBatches.incrementAndGet();
        }, executor);                                 // ★ 用有界线程池
        futures.add(future);
    }

    // 限时等待所有批次完成
    CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
                     .get(60, TimeUnit.MINUTES);     // ★ 超时兜底
} finally {
    gracefulShutdown(executor, 60, TimeUnit.MINUTES); // ★ 优雅停机
}

4.6 关键点

✅ ArrayBlockingQueue 严格有界，比 LinkedBlockingQueue 更安全
✅ ThreadFactoryBuilder 命名 + 守护 + 全局异常处理
✅ 固定大小线程池（core == max）适合导入校验
✅ CompletableFuture + 超时避免永久阻塞
✅ 优雅停机：shutdown() → 等超时 → shutdownNow()

💡 金句：线程池不是 Executors.newFixedThreadPool() 一行搞定，7 个参数每个都有讲究。

五、精讲 3：CallerRunsPolicy ⭐ 重头

5.1 一句话定义

CallerRunsPolicy = 队列满时，由提交任务的线程自己执行任务。

5.2 4 种内置拒绝策略对比

策略	行为	适用场景
AbortPolicy	抛 `RejectedExecutionException`	⚠️ 默认，业务可能崩
CallerRunsPolicy	提交线程自己跑	✅ 业务型（导入 / 提交流程）
DiscardPolicy	悄悄丢弃	⚠️ 通知型（丢一条不要紧）
DiscardOldestPolicy	丢最老的任务	⚠️ 看场景

5.3 CallerRunsPolicy 的反压语义

核心效果：上游"提交者"自动降速 → 自然限流。

5.4 项目里的"业务型 vs 通知型"分类

业务类型	选择	理由
导入校验	CallerRunsPolicy	丢一行整批要回滚
合同提交	CallerRunsPolicy	关键业务不能丢
通用业务池	CallerRunsPolicy	团队约定
机器人通知	DiscardPolicy	丢一条不阻塞主流程

5.5 为什么业务型选 CallerRunsPolicy？

理由	解释
✅ 不丢任务	即使队列满，任务也会执行（虽然慢）
✅ 天然限流	上游自动降速
✅ 失败友好	CallerRuns 让提交者同步执行，错误能立刻被发现
❌ 副作用	调用者线程会被卡住（所以上游要有超时控制）

5.6 为什么通知型选 DiscardPolicy？

理由	解释
✅ 不阻塞	丢一条消息比阻塞主流程划算
✅ 业务容忍	通知类业务允许少量丢失（用户没收到会再发）
❌ 副作用	任务悄悄没了，监控必须补上

5.7 关键点

✅ 业务型 → CallerRunsPolicy（项目团队约定）
✅ 通知型 → DiscardPolicy
❌ 不要无脑用 AbortPolicy（默认是它，但容易崩）
✅ CallerRunsPolicy 的"提交者卡住"是特性不是 bug

💷 金句：CallerRunsPolicy = “不让任务丢，让提交者慢”。通知型才用 DiscardPolicy。

六、3 大组件的关系图

金句：SAX 解决"读"，线程池解决"算"，CallerRunsPolicy 解决"稳"——三件套各司其职，合起来才是完整的 Excel 导入方案。

七、5 个常见踩坑（过来人的血泪教训）

#	踩坑	现象	解法
1	`EasyExcel.read()` 链式	5w 行 OOM	用 SAX 流式
2	`LinkedBlockingQueue()` 无参	任务堆积 100w	必须指定容量
3	默认 AbortPolicy	任务一多就抛异常	业务型改 CallerRunsPolicy
4	`Executors.newFixedThreadPool`	线程名 `pool-1-thread-1`，看不出业务	用 `ThreadFactoryBuilder` 命名
5	不优雅停机	发版时任务丢一半	`waitForTasksToCompleteOnShutdown=true`