EasyExcel 批量导入数据库,真正值钱的不是那几行 API

一份几万行的 Excel 上传上来，接口迟迟不返回，应用内存一路往上涨，数据库里还能看到密密麻麻的单条 INSERT。

这种导入代码我一般不用往下细看，大概率是两个问题：Excel 一次性全读进内存，数据再一行一行写数据库。

EasyExcel 解决的主要是前一个问题。

不过这里得把技术栈说清楚：EasyExcel 是 Java 的 Excel 处理工具，Python 项目不能直接安装使用。它的官方用法是通过 ReadListener 逐行接收数据，缓存到一定数量后批量写入数据库，最后再处理不足一个批次的剩余数据。官方示例本身也是这个思路。

所以，EasyExcel 真正值得拿过来的不是 API，而是这条处理链路：

逐行读取
    ↓
字段校验
    ↓
缓存一个批次
    ↓
批量写入临时表
    ↓
校验通过后合并正式表

我不太建议把几万行数据塞进一个 List，全部检查完再统一保存。Excel 稍微大一点，内存先扛不住。更别在监听器里每来一行就执行一次 SQL，那样 Excel 读取倒是快了，数据库又被打成了串行写入。

Python 里可以用 openpyxl 的只读模式实现同样的流式处理。这个模式会延迟读取工作表内容，不需要把整个工作簿展开在内存里。

下面这段是我更愿意放进项目里的写法，以 PostgreSQL 和 SQLAlchemy 为例：

from pathlib import Path
from uuid import uuid4

from openpyxl import load_workbook
from sqlalchemy import MetaData, Table, create_engine, text

BATCH_SIZE = 800
EXPECTED_HEADER = ("姓名", "手机号", "部门")


defflush_rows(engine, stage_table, rows):
ifnot rows:
return

with engine.begin() as conn:
        conn.execute(stage_table.insert(), rows)

    rows.clear()


defimport_users(file_path: str, database_url: str) -> dict:
    path = Path(file_path)
if path.suffix.lower() != ".xlsx":
raise ValueError("只接收 xlsx 文件")

    engine = create_engine(database_url, pool_pre_ping=True)
    metadata = MetaData()
    stage_table = Table(
"user_import_stage",
        metadata,
        autoload_with=engine,
    )

    task_id = uuid4().hex
    pending = []
    errors = []

    workbook = load_workbook(
        path,
        read_only=True,
        data_only=True,
    )

try:
        sheet = workbook["用户"]
        rows = sheet.iter_rows(values_only=True)

        header = tuple(
            str(value).strip() if value isnotNoneelse""
for value in next(rows)
        )

if header != EXPECTED_HEADER:
raise ValueError(f"表头不匹配，实际表头：{header}")

for row_number, row in enumerate(rows, start=2):
            name, mobile, department = row

            name = str(name).strip() if name else""
            mobile = str(mobile).strip() if mobile else""
            department = str(department).strip() if department else""

ifnot name andnot mobile:
continue

ifnot name:
                errors.append(f"第 {row_number} 行：姓名为空")
continue

ifnot mobile.isdigit() or len(mobile) != 11:
                errors.append(f"第 {row_number} 行：手机号格式错误")
continue

            pending.append({
"task_id": task_id,
"source_row": row_number,
"user_name": name,
"mobile": mobile,
"department": department,
            })

if len(pending) >= BATCH_SIZE:
                flush_rows(engine, stage_table, pending)

        flush_rows(engine, stage_table, pending)

finally:
        workbook.close()

if errors:
return {
"task_id": task_id,
"success": False,
"errors": errors[:100],
        }

with engine.begin() as conn:
        conn.execute(
            text("""
                INSERT INTO user_account
                    (user_name, mobile, department)
                SELECT user_name, mobile, department
                FROM user_import_stage
                WHERE task_id = :task_id
                ON CONFLICT (mobile) DO UPDATE SET
                    user_name = EXCLUDED.user_name,
                    department = EXCLUDED.department
            """),
            {"task_id": task_id},
        )

        conn.execute(
            text("""
                DELETE FROM user_import_stage
                WHERE task_id = :task_id
            """),
            {"task_id": task_id},
        )

return {
"task_id": task_id,
"success": True,
    }

这里没有把数据库事务从第一行一直开到最后一行，而是分批写入临时表。所有数据校验通过后，再一次性合并到正式表。

这么做稍微麻烦一点，但出了问题好处理。某一行手机号格式不对，不会把前面已经写入正式表的数据留下一半；任务重试时，也可以根据 task_id 清理临时数据。SQLAlchemy 接收参数列表时会走批量执行，而不是让业务代码手动循环执行单条插入。

EasyExcel 的优势也就在这里：逐行回调、对象映射、格式转换、异常处理和多 Sheet 读取都已经封装好了。Java 项目做大文件导入，代码确实会比直接操作 POI 清楚不少。

但低内存不等于导入一定快。

数据库还在单条插入、索引重复校验、事务一直不提交，换什么 Excel 库都救不了。批量导入真正要盯的是内存、批次大小、事务边界、错误回执和重复导入，而不是只看文件有没有成功读出来。

还有一点需要留意：EasyExcel 的官方 GitHub 仓库已经在 2025 年 9 月归档并转为只读。老项目继续使用问题不大，新项目准备引入时，版本固定、漏洞修复和后续维护都要提前评估。

Excel 只是入口。导入功能最后稳不稳，看的还是后面那段数据库处理。