1. 为什么要单独搞一个异步任务中心1.1 一次 30 万行导入把接口拖垮的现场先说结论批量导入这类活儿天生就不该塞进同步 HTTP 请求里。我最早接手的一个后台系统用户上传 Excel 导入客户资料接口里直接从MultipartFile读流、逐行校验、逐行 insert一个小文件几千行跑得挺欢产品经理也就默认这事儿成了。直到某天运营侧丢进来一个 30 万行的表浏览器转了二十多秒网关先超时返回 504用户以为失败又点了一次结果后台其实还在跑数据被插了两遍。更难受的是那段时间数据库连接池被长事务占满整个系统的其他查询全跟着卡。这个事故之后我就把导入这件事拆成了两半前台只负责把文件收下来、登记一条任务、立刻返回任务号真正的解析和入库交给后台的异步任务中心去跑。用户在页面上看到的是一个进度条可以关掉页面、可以去干别的回来再看结果。这套东西后来被我抽象成一个通用的模块凡是耗时超过两三秒、又不需要用户盯着同步等结果的操作统统往里丢。我这里说的简易异步任务中心重点在简易两个字。它不是什么大厂级别的工作流引擎不需要支持 DAG 编排、不需要跨机房容灾就是一张任务表加一个线程池加一套状态机一个中等规模的业务系统两三天就能落地。适合谁来参考后端开发、全栈、以及那些被迫兼着做后台管理系统的同学。如果你手上正有一个上传大文件然后处理的需求或者有一堆导出、报表生成、批量推送的活儿散落在各个接口里这套方案基本能直接抄。1.2 异步任务中心到底替你扛了什么事很多人对它的理解停留在把活儿挪到后台跑这个理解太浅。真正值钱的是它顺手解决了四个问题。第一是超时问题。HTTP 请求有生命周期网关有超时配置浏览器有耐心上限这三者都不可控。任务中心把长耗时操作从请求链路里切出去请求变成提交任务这种毫秒级操作超时自然消失。第二是可观测性。同步接口挂了就挂了用户报一句导入失败你连失败在第几行都不知道。有了任务表你能查到这次任务什么时候开始的、处理到第几行、哪几行校验没过、错误原因是什么甚至能把错误行导出来回给用户让他改完再传。第三是可控性。任务在跑的时候能不能取消能不能限流能不能在系统压力大的时候暂停同步请求里这些统统做不到任务中心里就是一个状态字段的事。第四是幂等与重试。用户手抖点两次提交、网络抖动导致客户端重发、节点重启导致任务中断这些问题在没有任务中心的时候要么丢数据要么重数据。有了任务表和唯一键重复提交直接被拦中断的任务能被扫描出来重新捡起来跑。我个人的判断标准很简单一个操作的耗时上限不可预估、或者可能超过 3 秒就必须异步化。别想着我这个文件最多也就几千行业务方永远会给你惊喜。1.3 什么规模的系统适合简易方案这里得把话说清楚免得有人照着做一个然后被流量打脸。单机 QPS 在几百以内、日任务量在几千到几万条、任务平均耗时在秒级到分钟级的系统数据库任务表 本地线程池完全够用运维成本几乎为零。上了十万级日任务量或者任务必须严格串行、需要多节点分摊你就该考虑引入正经的消息队列了。再往上任务之间有依赖关系、需要可视化编排那就是工作流引擎的领域了。我的经验是先用最简单的方案跑通业务等真的压不住了再升级。提前上重型组件最后大概率是团队里没人能维护出问题只能重启。简易方案的好处是逻辑全在你自己代码里出问题打开数据库就能看明白。2. 整体设计与技术选型的取舍2.1 三层结构提交层、调度层、执行层整套东西我一般拆成三块职责边界划清楚以后代码就好写了。提交层负责接收请求、做参数校验、算幂等键、落一条待执行的任务记录然后立刻返回任务号。这一层绝对不碰业务逻辑也不做任何耗时操作它的唯一 KPI 就是快。调度层是个定时任务每隔几秒扫一次任务表把待执行且没被别人抢占的任务捞出来扔进线程池。同时它还要负责另一件事扫描那些执行中但心跳已经超时的任务判定为节点异常改回待执行或者标记失败。执行层是真正干活的线程按业务类型分发到对应的处理器边处理边更新进度结束后写结果、写明细。三层之间只通过任务表通信没有直接调用。这个设计的好处是每一层都可以独立重启执行层的机器挂了不影响提交调度层停了任务就在表里堆着起来接着跑。2.2 线程池、消息队列还是纯数据库轮询这是被问得最多的一个问题我把几种常见组合的实测感受列一下。方案落地成本吞吐能力运维复杂度适用场景定时扫表 本地线程池极低中等极低单机、日任务量万级以内提交时直接丢线程池极低中等极低任务允许丢、不要求持久化扫表 消息队列投递中高中多节点、任务量大、要求可靠纯消息队列无任务表中高中只需执行、不关心进度查询我为什么坚持任务表是必需的哪怕你用了消息队列因为用户要查进度、要查失败原因、运营要统计每天导入了多少条这些查询需求消息队列满足不了。任务表是事实来源队列只是投递通道。至于提交时直接丢线程池这种写法看着最省事实际上最坑应用一重启队列里的任务全没了用户那边还以为在跑。除非这个任务丢了也无所谓否则别这么干。2.3 任务表设计状态机是整套方案的心脏表结构我直接贴出来字段不多但每一个都是我踩过坑之后加的。CREATE TABLE async_task ( id BIGINT NOT NULL AUTO_INCREMENT, task_no VARCHAR(64) NOT NULL COMMENT 对外任务号前端拿这个查进度, biz_type VARCHAR(32) NOT NULL COMMENT 业务类型如 IMPORT_CUSTOMER, biz_key VARCHAR(128) DEFAULT NULL COMMENT 业务幂等键通常用文件MD5, status TINYINT NOT NULL DEFAULT 0 COMMENT 0待执行 1执行中 2成功 3失败 4部分成功 5已取消, progress INT NOT NULL DEFAULT 0 COMMENT 0-100, total_count INT DEFAULT 0 COMMENT 总条数, success_count INT DEFAULT 0, fail_count INT DEFAULT 0, param_json TEXT COMMENT 任务参数文件地址、业务入参等, result_json TEXT COMMENT 结果摘要如错误文件下载地址, error_msg VARCHAR(1000) COMMENT 失败原因截断存储, retry_count INT NOT NULL DEFAULT 0, max_retry INT NOT NULL DEFAULT 3, exec_node VARCHAR(64) COMMENT 当前执行节点标识用于防并发, heartbeat_at DATETIME COMMENT 心跳时间判活用, created_by VARCHAR(64), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, started_at DATETIME DEFAULT NULL, finished_at DATETIME DEFAULT NULL, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_task_no (task_no), UNIQUE KEY uk_biz (biz_type, biz_key), KEY idx_status_created (status, created_at), KEY idx_heartbeat (status, heartbeat_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;几个字段值得单独说。uk_biz (biz_type, biz_key)这个唯一索引是幂等的第一道闸。用户重复上传同一个文件第二次插入直接报唯一键冲突接口捕获后返回该文件已在处理中或已处理完成请勿重复提交顺手把已有任务号返回去。这比在代码里先 select 再 insert 靠谱得多并发情况下不会漏。exec_node配合乐观锁更新保证同一时刻只有一个节点在跑这条任务UPDATE async_task SET status 1, exec_node #{nodeId}, started_at NOW(), heartbeat_at NOW() WHERE id #{id} AND status 0;影响行数为 1 才算抢到为 0 就跳过。别小看这一句多节点部署的时候它就是防止任务被执行两遍的关键。heartbeat_at是判活用的。执行中的任务每处理一批就更新一次心跳调度层扫到status1 AND heartbeat_at NOW() - INTERVAL 5 MINUTE的记录说明跑它的那个节点大概率挂了把任务重置回待执行并让retry_count 1。5 分钟这个值要根据你的单批处理耗时来定如果你的单批处理可能要 3 分钟那就设成 10 分钟不然会把正在跑的任务误判成僵尸。2.4 为什么我要拆成主表和明细表有人图省事把所有错误信息塞进error_msg一个字段里拼成一大段 JSON。小文件没问题三十万行里错五千行的时候这个字段直接爆掉而且用户根本没法定位。所以明细表是必须的CREATE TABLE async_task_item ( id BIGINT NOT NULL AUTO_INCREMENT, task_id BIGINT NOT NULL, batch_no INT NOT NULL COMMENT 批次号断点续传用, row_no INT NOT NULL COMMENT 文件中的原始行号, status TINYINT NOT NULL DEFAULT 0 COMMENT 0待处理 1成功 2失败, raw_data TEXT COMMENT 原始行内容失败时留存便于排查, error_field VARCHAR(128) COMMENT 出错字段, error_msg VARCHAR(500) COMMENT 出错原因, PRIMARY KEY (id), KEY idx_task_status (task_id, status), KEY idx_task_row (task_id, row_no) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;明细表我只记失败的行成功的行不落库。原因很现实三十万行全记这张表一个月就上千万查询慢、归档烦、磁盘还疼。失败行通常只占个位数百分比记下来足够定位问题也足够导出错误文件回给用户。如果业务要求每一行都要能追溯那就在明细表里记全量但一定要配task_id的分区或者定期归档策略别让它无限长。3. 核心代码从建表到跑通第一条任务3.1 提交入口怎么写才够稳提交接口的逻辑我固定成这个顺序一步都不能少校验文件类型、大小上限这个上限别设太宽松后面解析内存跟它直接相关计算文件 MD5把文件存到临时目录或者对象存储拿到可访问路径用biz_type MD5插入任务记录冲突则返回已有任务号返回任务号前端跳转到任务详情页开始轮询PostMapping(/import/submit) public ResultTaskVO submit(RequestParam(file) MultipartFile file) { String md5 DigestUtils.md5Hex(file.getInputStream()); String path storage.save(file, md5); // 落盘按 md5 命名天然去重 AsyncTask task new AsyncTask(); task.setTaskNo(IdGen.next(T)); task.setBizType(IMPORT_CUSTOMER); task.setBizKey(md5); task.setParamJson(JSON.toJSONString(Map.of(filePath, path))); task.setStatus(0); try { taskMapper.insert(task); } catch (DuplicateKeyException e) { AsyncTask exist taskMapper.selectByBizKey(IMPORT_CUSTOMER, md5); return Result.ok(TaskVO.of(exist).setDuplicate(true)); } return Result.ok(TaskVO.of(task)); }这里有个细节文件按 MD5 命名存储。好处是同一个文件重复上传不会占两份空间而且后续如果要做相同文件复跑会很方便。记得在文件清理任务里排除还在被任务引用的文件别把正在跑的任务的源文件删了。另外MultipartFile读流算 MD5 之后流的指针已经到末尾了如果后面还要读文件内容必须重新打开。这个坑我踩过一次现象是导入零条数据但任务显示成功查了半天才发现流被读空了。3.2 调度层扫表、抢锁、投递调度用的是 Spring 的Scheduled固定间隔 5 秒扫一次每次最多捞 20 条防止一次把所有任务都抢过来把自己的线程池撑爆。Scheduled(fixedDelay 5000) public void dispatch() { ListAsyncTask tasks taskMapper.selectPending(20); for (AsyncTask t : tasks) { int affected taskMapper.tryLock(t.getId(), nodeId); if (affected 0) continue; // 被别人抢走了 try { executor.execute(() - runSafely(t)); } catch (RejectedExecutionException e) { // 线程池满了退回待执行等下一轮 taskMapper.resetToPending(t.getId(), 线程池饱和等待重试); } } }selectPending的 SQL 要加索引ORDER BY created_at配合idx_status_created别让它变成全表扫。任务表膨胀到几百万行的时候这一句的性能差异非常明显。RejectedExecutionException的兜底处理特别重要。线程池满了以后如果不接住这个异常任务就会卡在执行中状态然后被心跳超时机制捡回来重跑白白浪费一轮时间。我现在统一的做法是拒绝就把状态退回 0下一轮再来同时打一条告警日志看到告警就去调大线程池或者查是不是有任务卡死了。线程池的配置我是这么定的Bean(importExecutor) public ThreadPoolExecutor importExecutor() { int core Math.max(4, Runtime.getRuntime().availableProcessors() / 2); return new ThreadPoolExecutor( core, core * 2, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue(200), new ThreadFactoryBuilder().setNameFormat(import-worker-%d).build(), new ThreadPoolExecutor.AbortPolicy() ); }core 不能按 CPU 核数满打满算导入是 IO 密集型任务但也不能开太大因为每个任务占着一个数据库连接和一份内存。我一般按核数的一半起步压测之后再调。队列设 200 是有讲究的队列太长任务在队列里等的时间不可控用户看到的排队中状态会拖很久队列太短稍微一波动就被拒绝。200 这个量级对中小系统来说是个比较舒服的中间值。3.3 执行器模板方法把公共逻辑收口每种任务的差异其实只在怎么处理这一段其他像状态更新、异常兜底、心跳、结果落库这些全部一样。所以我用模板方法把骨架固定下来。public abstract class AbstractTaskHandlerT { public void execute(AsyncTask task) { long begin System.currentTimeMillis(); try { taskMapper.updateStatus(task.getId(), 1); T param parseParam(task); TaskResult result doExecute(task, param); taskMapper.finish(task.getId(), result); } catch (BizException e) { // 业务异常不重试直接失败 taskMapper.fail(task.getId(), e.getMessage(), false); } catch (Exception e) { // 系统异常看次数决定是否重试 boolean canRetry task.getRetryCount() task.getMaxRetry(); taskMapper.fail(task.getId(), e.getMessage(), canRetry); } finally { log.info(task{} cost{}ms, task.getTaskNo(), System.currentTimeMillis() - begin); } } protected abstract TaskResult doExecute(AsyncTask task, T param); protected abstract T parseParam(AsyncTask task); }这里面最关键的是区分业务异常和系统异常。业务异常是指文件格式不对必填字段为空这类把任务重跑一万遍结果都一样重试纯属浪费资源。系统异常是指数据库连接超时锁等待超时这类值得重试。我在BizException上做了统一收口所有校验类错误都往上抛这个处理就是干脆利落的一次性失败。3.4 进度上报别每一行都 update进度更新如果每处理一行就 update 一次数据库三十万行就是三十万次写任务跑完数据库也快废了。我的做法是按批次更新并且做节流。private void reportProgress(Long taskId, int processed, int total) { int percent total 0 ? 0 : (int) (processed * 100L / total); // 百分比没变化就不写库 if (percent lastReported) return; taskMapper.updateProgress(taskId, percent, processed); lastReported percent; }再加一层每 500 行处理一次最多产生 200 次更新对数据库完全无压力。如果任务特别长还可以加个时间维度比如两次更新间隔至少 1 秒进一步压缩写入次数。前端轮询就简单了每 2 秒查一次任务详情进度到 100 或者状态变成终态就停下。别用 WebSocket为了一个进度条引入长连接运维和调试成本不成比例。轮询加个 2 秒间隔用户体感上完全够。4. 批量导入的解析、校验与入库4.1 流式读取内存杀手就在这一行EasyExcel 也好POI 也好最要命的写法是把整个文件读成一个List// 千万别这么写 ListCustomerRow all EasyExcel.read(file).head(CustomerRow.class).sheet().doReadSync();三十万行的对象列表每个对象算 200 字节光对象就是 60MB加上 Excel 解析过程中的中间对象翻个三五倍很轻松堆内存小的机器直接 OOM。而且这还只是三十万行一百万行呢正确姿势是用监听器分批回调处理完就释放EasyExcel.read(file, CustomerRow.class, new CustomerRowListener(importer, 500)) .sheet() .doRead();监听器里维护一个 500 大小的缓冲列表攒够就调一次业务处理然后clear()public class CustomerRowListener extends AnalysisEventListenerCustomerRow { private final ListCustomerRow buffer new ArrayList(BATCH_SIZE); private final Importer importer; Override public void invoke(CustomerRow row, AnalysisContext ctx) { row.setRowNo(ctx.readRowHolder().getRowIndex() 1); buffer.add(row); if (buffer.size() BATCH_SIZE) { importer.process(buffer); buffer.clear(); } } Override public void doAfterAllAnalysed(AnalysisContext ctx) { if (!buffer.isEmpty()) importer.process(buffer); } }BATCH_SIZE的取值我一般用 500偶尔用 1000。这个值和两个东西挂钩一是数据库单次批量 insert 的合理包大小二是占用的内存。500 条的 insert 语句大概几十 KB非常安全调大到 5000 有可能触发max_allowed_packet报错。还有一个隐藏的坑ctx.readRowHolder().getRowIndex()是从 0 开始的用户看到的是 Excel 里的行号从 1 开始而且带表头的话还得再加 1。错误明细回给用户的时候行号对不上用户根本找不到是哪一行这个细节必须处理好。4.2 事务边界整批一个事务是自杀早期我图省事让整个导入跑在一个事务里要么全成功要么全失败多干净。结果三十万行跑到 25 万行的时候一条数据违反约束整个事务回滚前面跑了五分钟全白费。更糟的是长事务把 undo log 撑得老大数据库性能肉眼可见地下降。现在的原则是批次级事务不是任务级事务。Transactional(propagation Propagation.REQUIRES_NEW) public void processBatch(ListCustomerRow rows, Long taskId) { ListCustomerRow valid new ArrayList(); ListAsyncTaskItem errors new ArrayList(); for (CustomerRow row : rows) { String err validator.check(row); if (err null) valid.add(row); else errors.add(buildErrorItem(taskId, row, err)); } if (!valid.isEmpty()) customerMapper.batchInsert(valid); if (!errors.isEmpty()) taskItemMapper.batchInsert(errors); counter.addSuccess(valid.size()); counter.addFail(errors.size()); }500 条一个事务失败只影响这一批其他批次照常推进进度也能正常更新。配合明细表记录失败行最后任务状态根据成功/失败比例给成功还是部分成功。这里有个必须注意的点Transactional如果在同一个类内部被自己调用this 调用代理不生效事务是假的。这个坑太多了我现在的做法是把批处理方法单独抽一个 Bean或者干脆用TransactionTemplate显式控制看得见摸得着。4.3 校验分层前置拦掉最便宜的错校验我分三层成本从低到高坏数据尽量在便宜的那一层就被拦掉。第一层是文件级校验在解析之前做文件头对不对、表头列名是否符合预期、行数是否超过上限、文件大小是否超限。这一层花不了几毫秒能挡掉一大半低级错误。第二层是行级格式校验解析时顺手做必填字段是否为空、手机号格式、日期格式、数字范围、枚举值是否合法。纯内存操作很快。第三层是业务校验需要查库客户编号是否已存在、外键引用是否存在、额度是否超限。这一层最贵所以我一定会把它放在格式校验之后并且批量查而不是逐行查。// 逐行查N 次数据库往返 for (CustomerRow row : rows) { if (customerMapper.existsByCode(row.getCode())) { ... } } // 批量查1 次往返 SetString codes rows.stream().map(CustomerRow::getCode).collect(toSet()); SetString exists customerMapper.selectExistingCodes(codes);这个优化在批量场景下效果非常夸张。500 行逐行查假设每次 1ms就是 500ms批量查一次 3ms。三十万行下来差了将近五分钟。批量导入的性能瓶颈九成都在循环里访问数据库这件事上。4.4 幂等与去重用户重复传了怎么办幂等分两个层面。任务层面靠前面说的biz_key 文件MD5 唯一索引同一个文件重复提交直接挡住。数据层面靠业务唯一键。文件改了改又传上来MD5 变了任务能进去但数据可能重复。这时候就得看业务语义了如果是新增客户用客户编号做唯一约束重复的走更新upsert或者跳过并记入失败明细如果是全量覆盖先按批次打标跑完统一处理旧数据如果是追加流水那就得让业务方确认重复导入本身就该拒绝我倾向于优先用数据库唯一索引兜底代码里的判重逻辑在并发情况下不靠谱。插入时用INSERT ... ON DUPLICATE KEY UPDATE或者捕获DuplicateKeyException然后把这部分行记进失败明细理由写数据已存在。INSERT INTO customer (code, name, phone, created_at) VALUES (#{code}, #{name}, #{phone}, NOW()) ON DUPLICATE KEY UPDATE name VALUES(name), phone VALUES(phone), updated_at NOW();这里有个坑ON DUPLICATE KEY UPDATE在批量插入时返回的 affected rows 不等于实际插入行数更新算 2插入算 1如果拿这个数去算成功条数会算错。我现在统一用提交的条数作为成功数真实影响行数只用来判断有没有更新。4.5 大文件上传与临时文件管理文件别存在应用本地磁盘上尤其是多节点部署的时候——A 节点收的文件任务被 B 节点抢走了B 节点根本读不到。上对象存储是标准做法路径按业务类型/日期/MD5.ext组织。如果条件不允许必须在本地存那就得做节点亲和任务表里记一个file_node调度时只调度到有这份文件的节点。这个方案我不太推荐可用性太差节点一挂文件就丢了。临时文件清理也很重要。我配置了一个每天凌晨跑的清理任务扫本地临时目录里超过 7 天、且没有活动任务引用的文件删掉。没做这个之前磁盘被历史导入文件撑满过一次挺尴尬的。5. 失败重试、断点续传与错误回执5.1 重试要有边界别无限重试我把重试分成三类处理异常类型举例是否重试处理方式业务校验失败手机号格式错、编号重复不重试记明细任务标记部分成功瞬时系统异常死锁、连接超时、锁等待重试指数退避最多 3 次不可恢复异常文件损坏、格式不支持不重试任务失败错误信息给用户重试的间隔我用的是 1 分钟、5 分钟、15 分钟这种递增避免一瞬间反复撞同一个问题。实现上不搞 sleep 阻塞线程而是在任务表加一个next_retry_at字段调度扫表时带上AND next_retry_at NOW()条件到点了才会被捞起来。还要有个最大重试次数的硬上限。见过一个系统重试没上限一条坏任务在那儿反复跑了一整晚把日志刷了几十 G。5.2 断点续传记录处理到哪一批了任务跑到一半节点挂了重启后从头再来一遍用户等了十分钟看到进度从 80% 退回 0%观感极差。解决办法是记批次号。在任务表加last_batch_no每批处理完跟进度一起更新。恢复时从批次表或者按last_batch_no * batchSize算出偏移量跳过。int startBatch task.getLastBatchNo() null ? 0 : task.getLastBatchNo(); listener.setSkipRows(startBatch * BATCH_SIZE);EasyExcel 支持跳过前 N 行headRowNumber和自定义跳过逻辑也可以自己数行号跳过。我用的是后者更可控Override public void invoke(CustomerRow row, AnalysisContext ctx) { int rowNo ctx.readRowHolder().getRowIndex() 1; if (rowNo skipRows) return; // 已处理过的跳过 // ... }注意断点续传的前提是每批处理是幂等的。否则重跑的那一批会插重复数据。这也是为什么我一直强调业务唯一键它是断点续传能成立的基础。5.3 错误回执把错误文件还给用户用户最想看到的不是导入失败而是第 128 行手机号字段格式不正确。所以任务结束后如果失败行数大于零我会生成一个错误文件原样保留原始数据在最右边加一列错误原因存到对象存储把下载链接放进result_json。public String exportErrorFile(Long taskId) { ListAsyncTaskItem errors taskItemMapper.selectByTaskId(taskId); String path error/ taskId .xlsx; EasyExcel.write(path, ErrorRow.class).sheet(错误明细).doWrite(errors); return storage.upload(path); }这个功能看起来是锦上添花实际上极大降低了客服压力。以前用户只会说导入没成功现在能自己对着错误文件改完重新传。另外错误信息别暴露内部细节。Duplicate entry xxx for key PRIMARY这种数据库原文直接抛给用户既不友好也不安全。统一转成数据已存在这类人话原始异常写到日志里给开发看。6. 常见问题排查实录6.1 任务一直卡在执行中这是最高频的问题原因通常有三个。节点进程崩了没更新状态。靠心跳超时机制兜底但要确认心跳确实在更新。我见过一个实现心跳只在任务开始时写一次任务跑二十分钟五分钟就被判死重置于是同一条任务被两个节点同时跑数据直接翻倍。心跳必须跟着批次走每批更新一次。线程被卡死了。典型场景是数据库锁等待、调用外部接口没设超时。给所有外部调用加超时是硬性要求别指望对方永远响应。事务没提交。长事务一直开着状态更新在同一个事务里看不到。检查方法很简单SHOW PROCESSLIST看一眼有没有长时间Sleep或者Query的连接。6.2 进度条卡在 99% 不动十有八九是总数估算不准。Excel 的总行数在很多实现里是估算值或者干脆用已处理批次 × 每批条数算出来的最后一批不满时进度就到不了 100。我的做法是解析前先快速扫一遍拿到准确行数EasyExcel 的AnalysisEventListener里可以通过context.readSheetHolder().getApproximateTotalRowNumber()拿但它是近似值。如果行数无法准确预知那就不显示百分比改成显示已处理 N 条用户反而更踏实。还有一种情况是最后一批处理完但状态更新的语句被吞了。加个兜底任务结束时强制把progress置成 100不管之前算出来是多少。6.3 同一条任务被执行了两次排查方向按这个顺序走抢锁 SQL 有没有带AND status 0条件抢锁之后有没有检查返回的影响行数心跳超时时间是不是设得太短把正在跑的任务误判成僵尸调度是不是部署了多个实例但每个实例都扫全表这个是对的抢锁是否可靠业务处理本身是否幂等第 3 条最容易忽略。假设你的单批处理要 5 分钟比如每批要调一次外部接口心跳间隔设成 5 分钟判活阈值设成 6 分钟那就可能出现批处理还没结束心跳还没到下一次更新任务被判死了。判活阈值至少要是心跳间隔的 3 倍。6.4 排查速查表现象可能原因快速验证处理任务堆在待执行不动调度未启动 / 线程池满看调度日志、线程池活跃数重启调度、扩线程池任务执行中状态不动心跳未更新 / 卡在外部调用查 heartbeat_at 时间补心跳逻辑、加超时导入数据重复幂等键失效 / 任务重复执行查 biz_key 唯一索引、exec_node补唯一索引、修抢锁内存持续上涨直至 OOM全量读文件 / 明细全量落库jmap 看大对象改流式读取、只记失败行数据库连接耗尽长事务 / 循环查库看连接池监控和慢查询缩事务、改批量查进度到 100 但数据不全最后一批未 flush查明细和实际数据量补 doAfterAllAnalysed 处理7. 监控、容量与上线前的自检7.1 必须盯住的几个指标任务中心这种东西平时没人看出事就是大事所以指标要提前埋好。积压量status 0的任务数超过 50 就该告警了说明消费跟不上。执行时长 P99同一业务类型任务的耗时分布某天突然变长大概率是数据量涨了或者下游变慢了。失败率按业务类型统计超过 10% 要人工介入看看。线程池活跃数与队列长度这是判断要不要扩容的直接依据我一般把队列使用率超过 80% 设成告警线。心跳超时重置次数这个数如果一直涨说明有节点不稳定得去查了。任务表的历史数据要定期归档。我的做法是保留最近 30 天的任务在热表更早的搬到async_task_history明细表跟着主表一起归档。不归档的话半年后这张表几百万行查个列表都要好几秒。7.2 上线前照着走一遍的清单任务表、明细表的索引都建了selectPending走的是idx_status_created抢锁 SQL 带了status 0条件并且检查了影响行数biz_type biz_key唯一索引存在重复提交会返回已有任务号心跳在每批处理时更新判活阈值是心跳间隔的 3 倍以上重试有上限指数退避业务异常不重试解析用流式内存占用和文件大小基本无关数据库操作全部批量循环里没有单条 SQL事务粒度是批次级没有跨批长事务失败明细落表错误文件可导出外部调用全部设置了超时文件存储路径按 MD5 组织有清理策略关键指标有监控和告警这套东西我前后在几个项目里用过从日任务量几百到几万都撑得住。踩得最狠的一次坑是刚开始没做心跳测试环境单机跑没问题上了生产两个节点一条任务被跑了四遍清理脏数据花了一晚上。从那以后我加了一条铁律任何涉及状态流转的字段更新都要先问一句并发情况下这行代码会怎样。异步任务中心的复杂性本质上全部来自于并发和故障把这两件事想透了剩下的都是体力活。
1. 为什么要单独搞一个异步任务中心1.1 一次 30 万行导入把接口拖垮的现场先说结论批量导入这类活儿天生就不该塞进同步 HTTP 请求里。我最早接手的一个后台系统用户上传 Excel 导入客户资料接口里直接从MultipartFile读流、逐行校验、逐行 insert一个小文件几千行跑得挺欢产品经理也就默认这事儿成了。直到某天运营侧丢进来一个 30 万行的表浏览器转了二十多秒网关先超时返回 504用户以为失败又点了一次结果后台其实还在跑数据被插了两遍。更难受的是那段时间数据库连接池被长事务占满整个系统的其他查询全跟着卡。这个事故之后我就把导入这件事拆成了两半前台只负责把文件收下来、登记一条任务、立刻返回任务号真正的解析和入库交给后台的异步任务中心去跑。用户在页面上看到的是一个进度条可以关掉页面、可以去干别的回来再看结果。这套东西后来被我抽象成一个通用的模块凡是耗时超过两三秒、又不需要用户盯着同步等结果的操作统统往里丢。我这里说的简易异步任务中心重点在简易两个字。它不是什么大厂级别的工作流引擎不需要支持 DAG 编排、不需要跨机房容灾就是一张任务表加一个线程池加一套状态机一个中等规模的业务系统两三天就能落地。适合谁来参考后端开发、全栈、以及那些被迫兼着做后台管理系统的同学。如果你手上正有一个上传大文件然后处理的需求或者有一堆导出、报表生成、批量推送的活儿散落在各个接口里这套方案基本能直接抄。1.2 异步任务中心到底替你扛了什么事很多人对它的理解停留在把活儿挪到后台跑这个理解太浅。真正值钱的是它顺手解决了四个问题。第一是超时问题。HTTP 请求有生命周期网关有超时配置浏览器有耐心上限这三者都不可控。任务中心把长耗时操作从请求链路里切出去请求变成提交任务这种毫秒级操作超时自然消失。第二是可观测性。同步接口挂了就挂了用户报一句导入失败你连失败在第几行都不知道。有了任务表你能查到这次任务什么时候开始的、处理到第几行、哪几行校验没过、错误原因是什么甚至能把错误行导出来回给用户让他改完再传。第三是可控性。任务在跑的时候能不能取消能不能限流能不能在系统压力大的时候暂停同步请求里这些统统做不到任务中心里就是一个状态字段的事。第四是幂等与重试。用户手抖点两次提交、网络抖动导致客户端重发、节点重启导致任务中断这些问题在没有任务中心的时候要么丢数据要么重数据。有了任务表和唯一键重复提交直接被拦中断的任务能被扫描出来重新捡起来跑。我个人的判断标准很简单一个操作的耗时上限不可预估、或者可能超过 3 秒就必须异步化。别想着我这个文件最多也就几千行业务方永远会给你惊喜。1.3 什么规模的系统适合简易方案这里得把话说清楚免得有人照着做一个然后被流量打脸。单机 QPS 在几百以内、日任务量在几千到几万条、任务平均耗时在秒级到分钟级的系统数据库任务表 本地线程池完全够用运维成本几乎为零。上了十万级日任务量或者任务必须严格串行、需要多节点分摊你就该考虑引入正经的消息队列了。再往上任务之间有依赖关系、需要可视化编排那就是工作流引擎的领域了。我的经验是先用最简单的方案跑通业务等真的压不住了再升级。提前上重型组件最后大概率是团队里没人能维护出问题只能重启。简易方案的好处是逻辑全在你自己代码里出问题打开数据库就能看明白。2. 整体设计与技术选型的取舍2.1 三层结构提交层、调度层、执行层整套东西我一般拆成三块职责边界划清楚以后代码就好写了。提交层负责接收请求、做参数校验、算幂等键、落一条待执行的任务记录然后立刻返回任务号。这一层绝对不碰业务逻辑也不做任何耗时操作它的唯一 KPI 就是快。调度层是个定时任务每隔几秒扫一次任务表把待执行且没被别人抢占的任务捞出来扔进线程池。同时它还要负责另一件事扫描那些执行中但心跳已经超时的任务判定为节点异常改回待执行或者标记失败。执行层是真正干活的线程按业务类型分发到对应的处理器边处理边更新进度结束后写结果、写明细。三层之间只通过任务表通信没有直接调用。这个设计的好处是每一层都可以独立重启执行层的机器挂了不影响提交调度层停了任务就在表里堆着起来接着跑。2.2 线程池、消息队列还是纯数据库轮询这是被问得最多的一个问题我把几种常见组合的实测感受列一下。方案落地成本吞吐能力运维复杂度适用场景定时扫表 本地线程池极低中等极低单机、日任务量万级以内提交时直接丢线程池极低中等极低任务允许丢、不要求持久化扫表 消息队列投递中高中多节点、任务量大、要求可靠纯消息队列无任务表中高中只需执行、不关心进度查询我为什么坚持任务表是必需的哪怕你用了消息队列因为用户要查进度、要查失败原因、运营要统计每天导入了多少条这些查询需求消息队列满足不了。任务表是事实来源队列只是投递通道。至于提交时直接丢线程池这种写法看着最省事实际上最坑应用一重启队列里的任务全没了用户那边还以为在跑。除非这个任务丢了也无所谓否则别这么干。2.3 任务表设计状态机是整套方案的心脏表结构我直接贴出来字段不多但每一个都是我踩过坑之后加的。CREATE TABLE async_task ( id BIGINT NOT NULL AUTO_INCREMENT, task_no VARCHAR(64) NOT NULL COMMENT 对外任务号前端拿这个查进度, biz_type VARCHAR(32) NOT NULL COMMENT 业务类型如 IMPORT_CUSTOMER, biz_key VARCHAR(128) DEFAULT NULL COMMENT 业务幂等键通常用文件MD5, status TINYINT NOT NULL DEFAULT 0 COMMENT 0待执行 1执行中 2成功 3失败 4部分成功 5已取消, progress INT NOT NULL DEFAULT 0 COMMENT 0-100, total_count INT DEFAULT 0 COMMENT 总条数, success_count INT DEFAULT 0, fail_count INT DEFAULT 0, param_json TEXT COMMENT 任务参数文件地址、业务入参等, result_json TEXT COMMENT 结果摘要如错误文件下载地址, error_msg VARCHAR(1000) COMMENT 失败原因截断存储, retry_count INT NOT NULL DEFAULT 0, max_retry INT NOT NULL DEFAULT 3, exec_node VARCHAR(64) COMMENT 当前执行节点标识用于防并发, heartbeat_at DATETIME COMMENT 心跳时间判活用, created_by VARCHAR(64), created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, started_at DATETIME DEFAULT NULL, finished_at DATETIME DEFAULT NULL, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (id), UNIQUE KEY uk_task_no (task_no), UNIQUE KEY uk_biz (biz_type, biz_key), KEY idx_status_created (status, created_at), KEY idx_heartbeat (status, heartbeat_at) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;几个字段值得单独说。uk_biz (biz_type, biz_key)这个唯一索引是幂等的第一道闸。用户重复上传同一个文件第二次插入直接报唯一键冲突接口捕获后返回该文件已在处理中或已处理完成请勿重复提交顺手把已有任务号返回去。这比在代码里先 select 再 insert 靠谱得多并发情况下不会漏。exec_node配合乐观锁更新保证同一时刻只有一个节点在跑这条任务UPDATE async_task SET status 1, exec_node #{nodeId}, started_at NOW(), heartbeat_at NOW() WHERE id #{id} AND status 0;影响行数为 1 才算抢到为 0 就跳过。别小看这一句多节点部署的时候它就是防止任务被执行两遍的关键。heartbeat_at是判活用的。执行中的任务每处理一批就更新一次心跳调度层扫到status1 AND heartbeat_at NOW() - INTERVAL 5 MINUTE的记录说明跑它的那个节点大概率挂了把任务重置回待执行并让retry_count 1。5 分钟这个值要根据你的单批处理耗时来定如果你的单批处理可能要 3 分钟那就设成 10 分钟不然会把正在跑的任务误判成僵尸。2.4 为什么我要拆成主表和明细表有人图省事把所有错误信息塞进error_msg一个字段里拼成一大段 JSON。小文件没问题三十万行里错五千行的时候这个字段直接爆掉而且用户根本没法定位。所以明细表是必须的CREATE TABLE async_task_item ( id BIGINT NOT NULL AUTO_INCREMENT, task_id BIGINT NOT NULL, batch_no INT NOT NULL COMMENT 批次号断点续传用, row_no INT NOT NULL COMMENT 文件中的原始行号, status TINYINT NOT NULL DEFAULT 0 COMMENT 0待处理 1成功 2失败, raw_data TEXT COMMENT 原始行内容失败时留存便于排查, error_field VARCHAR(128) COMMENT 出错字段, error_msg VARCHAR(500) COMMENT 出错原因, PRIMARY KEY (id), KEY idx_task_status (task_id, status), KEY idx_task_row (task_id, row_no) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;明细表我只记失败的行成功的行不落库。原因很现实三十万行全记这张表一个月就上千万查询慢、归档烦、磁盘还疼。失败行通常只占个位数百分比记下来足够定位问题也足够导出错误文件回给用户。如果业务要求每一行都要能追溯那就在明细表里记全量但一定要配task_id的分区或者定期归档策略别让它无限长。3. 核心代码从建表到跑通第一条任务3.1 提交入口怎么写才够稳提交接口的逻辑我固定成这个顺序一步都不能少校验文件类型、大小上限这个上限别设太宽松后面解析内存跟它直接相关计算文件 MD5把文件存到临时目录或者对象存储拿到可访问路径用biz_type MD5插入任务记录冲突则返回已有任务号返回任务号前端跳转到任务详情页开始轮询PostMapping(/import/submit) public ResultTaskVO submit(RequestParam(file) MultipartFile file) { String md5 DigestUtils.md5Hex(file.getInputStream()); String path storage.save(file, md5); // 落盘按 md5 命名天然去重 AsyncTask task new AsyncTask(); task.setTaskNo(IdGen.next(T)); task.setBizType(IMPORT_CUSTOMER); task.setBizKey(md5); task.setParamJson(JSON.toJSONString(Map.of(filePath, path))); task.setStatus(0); try { taskMapper.insert(task); } catch (DuplicateKeyException e) { AsyncTask exist taskMapper.selectByBizKey(IMPORT_CUSTOMER, md5); return Result.ok(TaskVO.of(exist).setDuplicate(true)); } return Result.ok(TaskVO.of(task)); }这里有个细节文件按 MD5 命名存储。好处是同一个文件重复上传不会占两份空间而且后续如果要做相同文件复跑会很方便。记得在文件清理任务里排除还在被任务引用的文件别把正在跑的任务的源文件删了。另外MultipartFile读流算 MD5 之后流的指针已经到末尾了如果后面还要读文件内容必须重新打开。这个坑我踩过一次现象是导入零条数据但任务显示成功查了半天才发现流被读空了。3.2 调度层扫表、抢锁、投递调度用的是 Spring 的Scheduled固定间隔 5 秒扫一次每次最多捞 20 条防止一次把所有任务都抢过来把自己的线程池撑爆。Scheduled(fixedDelay 5000) public void dispatch() { ListAsyncTask tasks taskMapper.selectPending(20); for (AsyncTask t : tasks) { int affected taskMapper.tryLock(t.getId(), nodeId); if (affected 0) continue; // 被别人抢走了 try { executor.execute(() - runSafely(t)); } catch (RejectedExecutionException e) { // 线程池满了退回待执行等下一轮 taskMapper.resetToPending(t.getId(), 线程池饱和等待重试); } } }selectPending的 SQL 要加索引ORDER BY created_at配合idx_status_created别让它变成全表扫。任务表膨胀到几百万行的时候这一句的性能差异非常明显。RejectedExecutionException的兜底处理特别重要。线程池满了以后如果不接住这个异常任务就会卡在执行中状态然后被心跳超时机制捡回来重跑白白浪费一轮时间。我现在统一的做法是拒绝就把状态退回 0下一轮再来同时打一条告警日志看到告警就去调大线程池或者查是不是有任务卡死了。线程池的配置我是这么定的Bean(importExecutor) public ThreadPoolExecutor importExecutor() { int core Math.max(4, Runtime.getRuntime().availableProcessors() / 2); return new ThreadPoolExecutor( core, core * 2, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue(200), new ThreadFactoryBuilder().setNameFormat(import-worker-%d).build(), new ThreadPoolExecutor.AbortPolicy() ); }core 不能按 CPU 核数满打满算导入是 IO 密集型任务但也不能开太大因为每个任务占着一个数据库连接和一份内存。我一般按核数的一半起步压测之后再调。队列设 200 是有讲究的队列太长任务在队列里等的时间不可控用户看到的排队中状态会拖很久队列太短稍微一波动就被拒绝。200 这个量级对中小系统来说是个比较舒服的中间值。3.3 执行器模板方法把公共逻辑收口每种任务的差异其实只在怎么处理这一段其他像状态更新、异常兜底、心跳、结果落库这些全部一样。所以我用模板方法把骨架固定下来。public abstract class AbstractTaskHandlerT { public void execute(AsyncTask task) { long begin System.currentTimeMillis(); try { taskMapper.updateStatus(task.getId(), 1); T param parseParam(task); TaskResult result doExecute(task, param); taskMapper.finish(task.getId(), result); } catch (BizException e) { // 业务异常不重试直接失败 taskMapper.fail(task.getId(), e.getMessage(), false); } catch (Exception e) { // 系统异常看次数决定是否重试 boolean canRetry task.getRetryCount() task.getMaxRetry(); taskMapper.fail(task.getId(), e.getMessage(), canRetry); } finally { log.info(task{} cost{}ms, task.getTaskNo(), System.currentTimeMillis() - begin); } } protected abstract TaskResult doExecute(AsyncTask task, T param); protected abstract T parseParam(AsyncTask task); }这里面最关键的是区分业务异常和系统异常。业务异常是指文件格式不对必填字段为空这类把任务重跑一万遍结果都一样重试纯属浪费资源。系统异常是指数据库连接超时锁等待超时这类值得重试。我在BizException上做了统一收口所有校验类错误都往上抛这个处理就是干脆利落的一次性失败。3.4 进度上报别每一行都 update进度更新如果每处理一行就 update 一次数据库三十万行就是三十万次写任务跑完数据库也快废了。我的做法是按批次更新并且做节流。private void reportProgress(Long taskId, int processed, int total) { int percent total 0 ? 0 : (int) (processed * 100L / total); // 百分比没变化就不写库 if (percent lastReported) return; taskMapper.updateProgress(taskId, percent, processed); lastReported percent; }再加一层每 500 行处理一次最多产生 200 次更新对数据库完全无压力。如果任务特别长还可以加个时间维度比如两次更新间隔至少 1 秒进一步压缩写入次数。前端轮询就简单了每 2 秒查一次任务详情进度到 100 或者状态变成终态就停下。别用 WebSocket为了一个进度条引入长连接运维和调试成本不成比例。轮询加个 2 秒间隔用户体感上完全够。4. 批量导入的解析、校验与入库4.1 流式读取内存杀手就在这一行EasyExcel 也好POI 也好最要命的写法是把整个文件读成一个List// 千万别这么写 ListCustomerRow all EasyExcel.read(file).head(CustomerRow.class).sheet().doReadSync();三十万行的对象列表每个对象算 200 字节光对象就是 60MB加上 Excel 解析过程中的中间对象翻个三五倍很轻松堆内存小的机器直接 OOM。而且这还只是三十万行一百万行呢正确姿势是用监听器分批回调处理完就释放EasyExcel.read(file, CustomerRow.class, new CustomerRowListener(importer, 500)) .sheet() .doRead();监听器里维护一个 500 大小的缓冲列表攒够就调一次业务处理然后clear()public class CustomerRowListener extends AnalysisEventListenerCustomerRow { private final ListCustomerRow buffer new ArrayList(BATCH_SIZE); private final Importer importer; Override public void invoke(CustomerRow row, AnalysisContext ctx) { row.setRowNo(ctx.readRowHolder().getRowIndex() 1); buffer.add(row); if (buffer.size() BATCH_SIZE) { importer.process(buffer); buffer.clear(); } } Override public void doAfterAllAnalysed(AnalysisContext ctx) { if (!buffer.isEmpty()) importer.process(buffer); } }BATCH_SIZE的取值我一般用 500偶尔用 1000。这个值和两个东西挂钩一是数据库单次批量 insert 的合理包大小二是占用的内存。500 条的 insert 语句大概几十 KB非常安全调大到 5000 有可能触发max_allowed_packet报错。还有一个隐藏的坑ctx.readRowHolder().getRowIndex()是从 0 开始的用户看到的是 Excel 里的行号从 1 开始而且带表头的话还得再加 1。错误明细回给用户的时候行号对不上用户根本找不到是哪一行这个细节必须处理好。4.2 事务边界整批一个事务是自杀早期我图省事让整个导入跑在一个事务里要么全成功要么全失败多干净。结果三十万行跑到 25 万行的时候一条数据违反约束整个事务回滚前面跑了五分钟全白费。更糟的是长事务把 undo log 撑得老大数据库性能肉眼可见地下降。现在的原则是批次级事务不是任务级事务。Transactional(propagation Propagation.REQUIRES_NEW) public void processBatch(ListCustomerRow rows, Long taskId) { ListCustomerRow valid new ArrayList(); ListAsyncTaskItem errors new ArrayList(); for (CustomerRow row : rows) { String err validator.check(row); if (err null) valid.add(row); else errors.add(buildErrorItem(taskId, row, err)); } if (!valid.isEmpty()) customerMapper.batchInsert(valid); if (!errors.isEmpty()) taskItemMapper.batchInsert(errors); counter.addSuccess(valid.size()); counter.addFail(errors.size()); }500 条一个事务失败只影响这一批其他批次照常推进进度也能正常更新。配合明细表记录失败行最后任务状态根据成功/失败比例给成功还是部分成功。这里有个必须注意的点Transactional如果在同一个类内部被自己调用this 调用代理不生效事务是假的。这个坑太多了我现在的做法是把批处理方法单独抽一个 Bean或者干脆用TransactionTemplate显式控制看得见摸得着。4.3 校验分层前置拦掉最便宜的错校验我分三层成本从低到高坏数据尽量在便宜的那一层就被拦掉。第一层是文件级校验在解析之前做文件头对不对、表头列名是否符合预期、行数是否超过上限、文件大小是否超限。这一层花不了几毫秒能挡掉一大半低级错误。第二层是行级格式校验解析时顺手做必填字段是否为空、手机号格式、日期格式、数字范围、枚举值是否合法。纯内存操作很快。第三层是业务校验需要查库客户编号是否已存在、外键引用是否存在、额度是否超限。这一层最贵所以我一定会把它放在格式校验之后并且批量查而不是逐行查。// 逐行查N 次数据库往返 for (CustomerRow row : rows) { if (customerMapper.existsByCode(row.getCode())) { ... } } // 批量查1 次往返 SetString codes rows.stream().map(CustomerRow::getCode).collect(toSet()); SetString exists customerMapper.selectExistingCodes(codes);这个优化在批量场景下效果非常夸张。500 行逐行查假设每次 1ms就是 500ms批量查一次 3ms。三十万行下来差了将近五分钟。批量导入的性能瓶颈九成都在循环里访问数据库这件事上。4.4 幂等与去重用户重复传了怎么办幂等分两个层面。任务层面靠前面说的biz_key 文件MD5 唯一索引同一个文件重复提交直接挡住。数据层面靠业务唯一键。文件改了改又传上来MD5 变了任务能进去但数据可能重复。这时候就得看业务语义了如果是新增客户用客户编号做唯一约束重复的走更新upsert或者跳过并记入失败明细如果是全量覆盖先按批次打标跑完统一处理旧数据如果是追加流水那就得让业务方确认重复导入本身就该拒绝我倾向于优先用数据库唯一索引兜底代码里的判重逻辑在并发情况下不靠谱。插入时用INSERT ... ON DUPLICATE KEY UPDATE或者捕获DuplicateKeyException然后把这部分行记进失败明细理由写数据已存在。INSERT INTO customer (code, name, phone, created_at) VALUES (#{code}, #{name}, #{phone}, NOW()) ON DUPLICATE KEY UPDATE name VALUES(name), phone VALUES(phone), updated_at NOW();这里有个坑ON DUPLICATE KEY UPDATE在批量插入时返回的 affected rows 不等于实际插入行数更新算 2插入算 1如果拿这个数去算成功条数会算错。我现在统一用提交的条数作为成功数真实影响行数只用来判断有没有更新。4.5 大文件上传与临时文件管理文件别存在应用本地磁盘上尤其是多节点部署的时候——A 节点收的文件任务被 B 节点抢走了B 节点根本读不到。上对象存储是标准做法路径按业务类型/日期/MD5.ext组织。如果条件不允许必须在本地存那就得做节点亲和任务表里记一个file_node调度时只调度到有这份文件的节点。这个方案我不太推荐可用性太差节点一挂文件就丢了。临时文件清理也很重要。我配置了一个每天凌晨跑的清理任务扫本地临时目录里超过 7 天、且没有活动任务引用的文件删掉。没做这个之前磁盘被历史导入文件撑满过一次挺尴尬的。5. 失败重试、断点续传与错误回执5.1 重试要有边界别无限重试我把重试分成三类处理异常类型举例是否重试处理方式业务校验失败手机号格式错、编号重复不重试记明细任务标记部分成功瞬时系统异常死锁、连接超时、锁等待重试指数退避最多 3 次不可恢复异常文件损坏、格式不支持不重试任务失败错误信息给用户重试的间隔我用的是 1 分钟、5 分钟、15 分钟这种递增避免一瞬间反复撞同一个问题。实现上不搞 sleep 阻塞线程而是在任务表加一个next_retry_at字段调度扫表时带上AND next_retry_at NOW()条件到点了才会被捞起来。还要有个最大重试次数的硬上限。见过一个系统重试没上限一条坏任务在那儿反复跑了一整晚把日志刷了几十 G。5.2 断点续传记录处理到哪一批了任务跑到一半节点挂了重启后从头再来一遍用户等了十分钟看到进度从 80% 退回 0%观感极差。解决办法是记批次号。在任务表加last_batch_no每批处理完跟进度一起更新。恢复时从批次表或者按last_batch_no * batchSize算出偏移量跳过。int startBatch task.getLastBatchNo() null ? 0 : task.getLastBatchNo(); listener.setSkipRows(startBatch * BATCH_SIZE);EasyExcel 支持跳过前 N 行headRowNumber和自定义跳过逻辑也可以自己数行号跳过。我用的是后者更可控Override public void invoke(CustomerRow row, AnalysisContext ctx) { int rowNo ctx.readRowHolder().getRowIndex() 1; if (rowNo skipRows) return; // 已处理过的跳过 // ... }注意断点续传的前提是每批处理是幂等的。否则重跑的那一批会插重复数据。这也是为什么我一直强调业务唯一键它是断点续传能成立的基础。5.3 错误回执把错误文件还给用户用户最想看到的不是导入失败而是第 128 行手机号字段格式不正确。所以任务结束后如果失败行数大于零我会生成一个错误文件原样保留原始数据在最右边加一列错误原因存到对象存储把下载链接放进result_json。public String exportErrorFile(Long taskId) { ListAsyncTaskItem errors taskItemMapper.selectByTaskId(taskId); String path error/ taskId .xlsx; EasyExcel.write(path, ErrorRow.class).sheet(错误明细).doWrite(errors); return storage.upload(path); }这个功能看起来是锦上添花实际上极大降低了客服压力。以前用户只会说导入没成功现在能自己对着错误文件改完重新传。另外错误信息别暴露内部细节。Duplicate entry xxx for key PRIMARY这种数据库原文直接抛给用户既不友好也不安全。统一转成数据已存在这类人话原始异常写到日志里给开发看。6. 常见问题排查实录6.1 任务一直卡在执行中这是最高频的问题原因通常有三个。节点进程崩了没更新状态。靠心跳超时机制兜底但要确认心跳确实在更新。我见过一个实现心跳只在任务开始时写一次任务跑二十分钟五分钟就被判死重置于是同一条任务被两个节点同时跑数据直接翻倍。心跳必须跟着批次走每批更新一次。线程被卡死了。典型场景是数据库锁等待、调用外部接口没设超时。给所有外部调用加超时是硬性要求别指望对方永远响应。事务没提交。长事务一直开着状态更新在同一个事务里看不到。检查方法很简单SHOW PROCESSLIST看一眼有没有长时间Sleep或者Query的连接。6.2 进度条卡在 99% 不动十有八九是总数估算不准。Excel 的总行数在很多实现里是估算值或者干脆用已处理批次 × 每批条数算出来的最后一批不满时进度就到不了 100。我的做法是解析前先快速扫一遍拿到准确行数EasyExcel 的AnalysisEventListener里可以通过context.readSheetHolder().getApproximateTotalRowNumber()拿但它是近似值。如果行数无法准确预知那就不显示百分比改成显示已处理 N 条用户反而更踏实。还有一种情况是最后一批处理完但状态更新的语句被吞了。加个兜底任务结束时强制把progress置成 100不管之前算出来是多少。6.3 同一条任务被执行了两次排查方向按这个顺序走抢锁 SQL 有没有带AND status 0条件抢锁之后有没有检查返回的影响行数心跳超时时间是不是设得太短把正在跑的任务误判成僵尸调度是不是部署了多个实例但每个实例都扫全表这个是对的抢锁是否可靠业务处理本身是否幂等第 3 条最容易忽略。假设你的单批处理要 5 分钟比如每批要调一次外部接口心跳间隔设成 5 分钟判活阈值设成 6 分钟那就可能出现批处理还没结束心跳还没到下一次更新任务被判死了。判活阈值至少要是心跳间隔的 3 倍。6.4 排查速查表现象可能原因快速验证处理任务堆在待执行不动调度未启动 / 线程池满看调度日志、线程池活跃数重启调度、扩线程池任务执行中状态不动心跳未更新 / 卡在外部调用查 heartbeat_at 时间补心跳逻辑、加超时导入数据重复幂等键失效 / 任务重复执行查 biz_key 唯一索引、exec_node补唯一索引、修抢锁内存持续上涨直至 OOM全量读文件 / 明细全量落库jmap 看大对象改流式读取、只记失败行数据库连接耗尽长事务 / 循环查库看连接池监控和慢查询缩事务、改批量查进度到 100 但数据不全最后一批未 flush查明细和实际数据量补 doAfterAllAnalysed 处理7. 监控、容量与上线前的自检7.1 必须盯住的几个指标任务中心这种东西平时没人看出事就是大事所以指标要提前埋好。积压量status 0的任务数超过 50 就该告警了说明消费跟不上。执行时长 P99同一业务类型任务的耗时分布某天突然变长大概率是数据量涨了或者下游变慢了。失败率按业务类型统计超过 10% 要人工介入看看。线程池活跃数与队列长度这是判断要不要扩容的直接依据我一般把队列使用率超过 80% 设成告警线。心跳超时重置次数这个数如果一直涨说明有节点不稳定得去查了。任务表的历史数据要定期归档。我的做法是保留最近 30 天的任务在热表更早的搬到async_task_history明细表跟着主表一起归档。不归档的话半年后这张表几百万行查个列表都要好几秒。7.2 上线前照着走一遍的清单任务表、明细表的索引都建了selectPending走的是idx_status_created抢锁 SQL 带了status 0条件并且检查了影响行数biz_type biz_key唯一索引存在重复提交会返回已有任务号心跳在每批处理时更新判活阈值是心跳间隔的 3 倍以上重试有上限指数退避业务异常不重试解析用流式内存占用和文件大小基本无关数据库操作全部批量循环里没有单条 SQL事务粒度是批次级没有跨批长事务失败明细落表错误文件可导出外部调用全部设置了超时文件存储路径按 MD5 组织有清理策略关键指标有监控和告警这套东西我前后在几个项目里用过从日任务量几百到几万都撑得住。踩得最狠的一次坑是刚开始没做心跳测试环境单机跑没问题上了生产两个节点一条任务被跑了四遍清理脏数据花了一晚上。从那以后我加了一条铁律任何涉及状态流转的字段更新都要先问一句并发情况下这行代码会怎样。异步任务中心的复杂性本质上全部来自于并发和故障把这两件事想透了剩下的都是体力活。
