Java如何快速处理百万数据Excel导入导出

/**
 * 🔹 技术亮点1：资源隔离 - 独立线程池处理导入导出，不影响核心业务
 * 🔹 技术亮点2：参数调优 - 基于cpu核心数和io密集型特点配置线程池
 */
@configuration
public class excelthreadpoolconfig {
    @bean("exceltaskexecutor")
    public threadpooltaskexecutor exceltaskexecutor() {
        threadpooltaskexecutor executor=new threadpooltaskexecutor();
        // io密集型：核心线程数=cpu核心数*2
        executor.setcorepoolsize(runtime.getruntime().availableprocessors()*2);
        executor.setmaxpoolsize(runtime.getruntime().availableprocessors()*4);
        executor.setqueuecapacity(1000);
        executor.setthreadnameprefix("excel-task-");
        // 拒绝策略：由调用线程执行，避免任务丢失
        executor.setrejectedexecutionhandler(new threadpoolexecutor.callerrunspolicy());
        executor.initialize();
        return executor;
    }
}

/**
 * 🔹 技术亮点3：流式读取 - easyexcel基于sax解析，内存占用<100mb
 * 🔹 技术亮点4：多线程分片 - 大文件按1000行分片，并行处理
 * 🔹 技术亮点5：jdbc真正批处理 - 开启rewritebatchedstatements参数
 */
@service
public class excelimportservice {
    @autowired
    private usermapper usermapper;
    @autowired
    @qualifier("exceltaskexecutor")
    private threadpooltaskexecutor executor;

    // 每批处理数据量（最佳值：1000-5000行）
    private static final int batch_size=2000;

    @async("exceltaskexecutor")
    public void importexcel(multipartfile file, long taskid) {
        try {
            // 初始化错误收集器
            list<excelerror> errorlist=collections.synchronizedlist(new arraylist<>());
            
            // easyexcel流式读取，逐行回调
            easyexcel.read(file.getinputstream(), userexceldto.class, new readlistener<userexceldto>() {
                // 线程安全的临时缓存
                private final list<userexceldto> cache=collections.synchronizedlist(new arraylist<>(batch_size));

                @override
                public void invoke(userexceldto data, analysiscontext context) {
                    // 1. 单条数据校验（jsr-380注解）
                    set<constraintviolation<userexceldto>> violations=validation.builddefaultvalidatorfactory().getvalidator().validate(data);
                    if (!violations.isempty()) {
                        // 记录错误行号和原因
                        integer rownum=context.readrowholder().getrowindex()+1;
                        string errormsg=violations.stream()
                                .map(constraintviolation::getmessage)
                                .collect(collectors.joining(", "));
                        errorlist.add(new excelerror(rownum, errormsg, data));
                        return;
                    }
                    
                    // 2. 加入缓存，达到批量阈值则提交处理
                    cache.add(data);
                    if (cache.size()>=batch_size) {
                        // 提交到线程池并行处理
                        executor.submit(() -> batchinsert(new arraylist<>(cache)));
                        cache.clear();
                    }
                }

                @override
                public void doafterallanalysed(analysiscontext context) {
                    // 处理剩余数据
                    if (!cache.isempty()) {
                        batchinsert(cache);
                    }
                    // 生成错误报告
                    if (!errorlist.isempty()) {
                        generateerrorreport(taskid, errorlist);
                    }
                    // 更新任务状态
                    updatetaskstatus(taskid, taskstatus.completed, errorlist.size());
                }
            }).sheet().doread();
        } catch (exception e) {
            updatetaskstatus(taskid, taskstatus.failed, 0);
            log.error("excel导入失败，任务id：{}", taskid, e);
        }
    }

    /**
     * 🔹 技术亮点6：批量插入优化 - 关闭自动提交+rewritebatchedstatements
     * 性能提升：100万条数据从30分钟→25秒
     */
    @transactional(rollbackfor=exception.class)
    public void batchinsert(list<userexceldto> list) {
        // 转换为do对象
        list<userdo> userdolist=list.stream()
                .map(this::converttodo)
                .collect(collectors.tolist());
        // mybatis批量插入（配合jdbc参数：rewritebatchedstatements=true）
        usermapper.batchinsert(userdolist);
    }
}

/**
 * 🔹 技术亮点7：流式写入 - sxssf将超过100行的数据写入临时文件
 * 🔹 技术亮点8：分批查询 - 查询一批写入一批，避免全量加载到内存
 */
@service
public class excelexportservice {
    @autowired
    private usermapper usermapper;
    @autowired
    private ossservice ossservice;

    private static final int batch_size=2000;
    // sxssf内存保留行数（默认100，可根据内存调整）
    private static final int window_size=100;

    @async("exceltaskexecutor")
    public void exportexcel(long taskid, userquerydto query) {
        sxssfworkbook workbook=null;
        try {
            // 初始化sxssf工作簿，开启流式写入
            workbook=new sxssfworkbook(window_size);
            sheet sheet=workbook.createsheet("用户数据");
            
            // 写入表头
            writeheader(sheet);
            
            // 游标分批查询数据库
            int offset=0;
            list<userdo> batch;
            do {
                // 每次查询2000条
                batch=usermapper.selectbypage(query, offset, batch_size);
                if (!batch.isempty()) {
                    // 写入当前批次数据
                    writedata(sheet, batch);
                    offset+=batch_size;
                }
            } while (!batch.isempty());
            
            // 上传到oss
            string filename="用户数据_"+system.currenttimemillis()+".xlsx";
            bytearrayoutputstream bos=new bytearrayoutputstream();
            workbook.write(bos);
            string downloadurl=ossservice.uploadfile(filename, bos.tobytearray());
            
            // 更新任务状态，发送下载链接
            updatetaskstatus(taskid, taskstatus.completed, downloadurl);
        } catch (exception e) {
            updatetaskstatus(taskid, taskstatus.failed, null);
            log.error("excel导出失败，任务id：{}", taskid, e);
        } finally {
            // 🔹 技术亮点9：清理临时文件 - 防止磁盘泄漏
            if (workbook!=null) {
                workbook.dispose();
            }
        }
    }
}

# application.yml
spring:
  datasource:
    url: jdbc:mysql://localhost:3306/test?useunicode=true&characterencoding=utf8&usessl=false
      &rewritebatchedstatements=true  # 🔹 必须开启！开启真正的jdbc批处理
      &cacheprepstmts=true
      &useserverprepstmts=true

@slf4j
@component
public class excelimportworker {

    // 每批处理行数，防止长事务
    private static final int batch_size = 1000;

    @autowired
    private jdbctemplate jdbctemplate;
    @autowired
    private importerrorservice errorservice;

    public void process(string filekey, long taskid) {
        // 从oss流式下载，直接对接easyexcel的inputstream
        try (inputstream inputstream = ossservice.getobject(filekey)) {
            list<importdto> batch = new arraylist<>(batch_size);
            
            easyexcel.read(inputstream, importdto.class, new readlistener<importdto>() {
                @override
                public void invoke(importdto data, analysiscontext context) {
                    // 1. 业务校验
                    list<string> errors = validate(data);
                    if (!errors.isempty()) {
                        // 校验失败，记录错误行
                        errorservice.saveerror(taskid, context.readrowholder().getrowindex(), data, errors);
                    } else {
                        batch.add(data);
                    }
                    // 2. 攒够一批，批量写入
                    if (batch.size() >= batch_size) {
                        batchinsert(batch);
                        batch.clear();
                        // 更新任务进度
                        taskservice.updateprogress(taskid, context.readrowholder().getrowindex());
                    }
                }

                @override
                public void doafterallanalysed(analysiscontext context) {
                    // 最后一批残留数据
                    if (!batch.isempty()) {
                        batchinsert(batch);
                    }
                    // 任务终态
                    taskservice.finish(taskid);
                }
            }).sheet().doread();

        } catch (exception e) {
            log.error("导入异常, taskid={}", taskid, e);
            taskservice.markfailed(taskid);
        }
    }

    // 亮点：jdbc batch + 手动事务，避免orm一级缓存膨胀
    @transactional(propagation = propagation.requires_new)
    public void batchinsert(list<importdto> list) {
        string sql = "insert into t_biz_data (...) values (?, ?, ...)";
        jdbctemplate.batchupdate(sql, new batchpreparedstatementsetter() {
            @override
            public void setvalues(preparedstatement ps, int i) throws sqlexception {
                importdto dto = list.get(i);
                ps.setstring(1, dto.getfield1());
                // ... 设置其他参数
            }
            @override
            public int getbatchsize() {
                return list.size();
            }
        });
    }
}

@slf4j
public class excelexportworker {

    private static final int query_page_size = 5000;

    public void export(long taskid, exportquery query) {
        // 直接获取oss上传的outputstream
        try (outputstream outputstream = ossservice.putobjectstream(exportkey)) {
            excelwriter excelwriter = easyexcel.write(outputstream, exportvo.class).build();
            writesheet writesheet = easyexcel.writersheet("数据页").build();

            long lastid = 0l;
            list<exportvo> pagelist;
            do {
                // 流式分页查询（避免全量拖入内存）
                pagelist = exportmapper.selectbypage(lastid, query_page_size, query);
                if (!pagelist.isempty()) {
                    excelwriter.write(pagelist, writesheet);
                    lastid = pagelist.get(pagelist.size() - 1).getid();
                }
                // 更新进度
                taskservice.updateprogress(taskid, lastid);
            } while (pagelist.size() == query_page_size);

            excelwriter.finish();
            taskservice.finish(taskid, exportkey);
        } catch (exception e) {
            log.error("导出异常, taskid={}", taskid, e);
            taskservice.markfailed(taskid);
        }
    }
}

@select("select * from t_biz_data where id > #{lastid} and ... order by id limit #{size}")
@options(resultsettype = resultsettype.forward_only, fetchsize = integer.min_value)
list<exportvo> selectbypage(@param("lastid") long lastid, @param("size") int size, @param("query") exportquery query);

@restcontroller
public class importcontroller {

    @postmapping("/import/upload")
    public result<long> upload(@requestparam("file") multipartfile file) {
        // 1. 文件直传oss
        string filekey = ossservice.upload(filetype.excel_import, file.getinputstream());
        // 2. 插入任务表
        long taskid = taskservice.createtask(tasktypeenum.import, filekey);
        // 3. 发送mq消息（解耦，异步化）
        rocketmqtemplate.asyncsend("excel-import-topic", new importmsg(taskid, filekey), null);
        return result.ok(taskid);
    }

    @getmapping("/import/progress/{taskid}")
    public result<taskprogressvo> progress(@pathvariable long taskid) {
        return result.ok(taskservice.getprogress(taskid));
    }
}