在后端系统中并不是所有任务都适合放在请求线程里同步执行。文件解析、图片处理、用户通知、报表生成、缓存刷新这类任务通常耗时较长并且不一定要阻塞当前请求。如果它们全部压在请求线程里接口响应会变慢请求线程也会被长时间占用。并发任务处理系统要解决的核心问题不只是“开几个线程异步执行任务”而是当任务不断产生而系统资源有限时如何让任务有边界地排队、被工作线程处理并在队列堆积、执行失败、系统关闭时保持可控。本章先讨论单机内存版任务处理系统。它不引入数据库任务表也不引入 MQ而是先把最小模型讲清楚任务如何建模为什么需要有界队列Worker 如何执行任务队列满了怎么办任务失败后如何重试以及系统如何监控和关闭。1、为什么不能在请求线程里直接处理任务假设有一个文件上传接口。用户上传文件后系统需要保存原文件、解析文件内容、生成预览图并通知用户处理结果。最直接的写法是publicvoidupload(Filefile){saveOriginalFile(file);parseFile(file);generatePreview(file);notifyUser(file);}这段代码的问题不是不能运行而是所有工作都压在请求线程里。只要文件变大、解析变慢、通知接口超时请求响应就会被拖慢。如果并发请求继续增加请求线程会越来越多地阻塞在这些耗时逻辑上最终可能导致 Web 容器的业务线程被占满。更合理的方式是让请求线程只完成必要的前置动作然后把后续任务交给后台系统处理publicvoidupload(Filefile){saveOriginalFile(file);TasktasknewTask(UUID.randomUUID().toString(),TaskType.FILE_PARSE,file.getId());taskSubmitter.submit(task);}这段代码里请求线程不再直接解析文件而是创建一个Task并提交出去。后续由后台 Worker 线程执行解析、生成预览和通知。这个变化的本质是请求线程负责产生任务后台 Worker 负责执行任务两者通过任务队列解耦。解耦之后请求线程不需要等待整个文件处理完成后台处理能力也可以通过 Worker 数量、队列容量、拒绝策略来单独控制。2、一个任务系统里有哪些关键角色一个单机并发任务处理系统最小可以拆成四个角色这几个角色分别承担不同职责角色作用在主例子中的对应Producer产生并提交任务文件上传接口Task描述一个待处理任务文件解析任务Task Queue暂存暂时处理不过来的任务有界阻塞队列Workers从队列中取任务并执行后台工作线程Processor封装具体业务逻辑解析文件、生成预览、通知用户这里要注意任务处理系统不是只有 Worker。Worker 只是执行者如果没有任务模型系统不知道自己执行的是什么如果没有队列任务提交速度和任务处理速度无法解耦如果没有 Processor业务逻辑会散落在 Worker 中后续扩展和测试都会变得困难。所以第一步不是急着调线程数而是先建立角色关系后面的设计都是围绕这条主线展开。3、为什么要把任务建模成 Task最简单的异步写法是直接提交一个Runnableexecutor.execute(()-{parseFile(fileId);generatePreview(fileId);notifyUser(fileId);});这种写法适合临时异步执行一段代码但不适合设计一个可管理的任务系统。因为Runnable只表达了“要执行什么代码”没有显式表达任务身份、任务类型、创建时间、重试次数和执行状态。一旦系统需要追踪任务属于谁、已经等待多久、失败过几次、当前执行到哪一步直接提交Runnable就不够了。因此任务系统通常需要先定义一个显式的任务对象importlombok.Getter;importlombok.RequiredArgsConstructor;GetterRequiredArgsConstructorpublicclassTask{privatefinalStringtaskId;privatefinalTaskTypetype;privatefinalStringbizId;privatefinallongcreateTimeSystem.currentTimeMillis();privateintretryCount0;privateTaskStatusstatusTaskStatus.CREATED;privateStringlastError;publicvoidincreaseRetryCount(){this.retryCount;}publicvoidmarkRunning(){this.statusTaskStatus.RUNNING;}publicvoidmarkSuccess(){this.statusTaskStatus.SUCCESS;}publicvoidmarkFailed(Stringerror){this.statusTaskStatus.FAILED;this.lastErrorerror;}publicvoidmarkDead(Stringerror){this.statusTaskStatus.DEAD;this.lastErrorerror;}}任务类型可以先包括FILE_PARSE、IMAGE_RESIZE、USER_NOTIFY任务状态可以先包括CREATED、RUNNING、SUCCESS、FAILED、DEAD。这里的bizId用来保存业务对象 ID。对于文件解析任务它可以是fileId对于用户通知任务它可以是userId对于订单同步任务它可以是orderId。这样设计之后任务不再只是一个代码片段而是一个可以被系统识别、排队、执行、监控和重试的对象。4、为什么任务要进入有界队列有了Task之后下一个问题是任务提交后是否应该立刻执行如果任务产生速度很慢Worker 又刚好空闲任务可以很快被执行。但系统设计不能只考虑空闲状态还要考虑任务突然增多的情况。比如某一段时间用户集中上传文件请求线程不断创建文件解析任务而 Worker 一时处理不过来这时就需要队列暂存任务。在单机内存模型中可以用BlockingQueueTask表示任务队列BlockingQueueTasktaskQueuenewArrayBlockingQueue(1000);这里使用ArrayBlockingQueue并且给定容量1000。重点不是具体数字而是队列必须有上限。如果使用无界队列BlockingQueueTasktaskQueuenewLinkedBlockingQueue();表面上任务不会被拒绝但压力只是被藏进内存里。只要任务提交速度长期大于处理速度队列就会持续增长最终带来内存压力、GC 压力甚至 OOM。所以队列不是为了无限缓存任务而是为了提供一个有限缓冲区。Worker 短时间处理不过来时任务可以先排队队列满了以后系统必须暴露过载信号而不是继续无边界接收任务。提交任务时推荐用offer()而不是put()publicclassTaskSubmitter{privatefinalBlockingQueueTasktaskQueue;publicTaskSubmitter(BlockingQueueTasktaskQueue){this.taskQueuetaskQueue;}publicSubmitResultsubmit(Tasktask){if(taskQueue.offer(task)){returnSubmitResult.ACCEPTED;}returnSubmitResult.REJECTED;}}提交结果可以定义为publicenumSubmitResult{ACCEPTED,REJECTED,CALLER_RUN,DROPPED}offer()的含义是如果队列有空间就放入任务如果队列已经满了马上返回false。这适合请求线程因为请求线程不能无限阻塞等待。与之相比put()在队列满时会一直阻塞。如果在请求线程里使用put()系统过载时请求线程会被越阻塞越多反而可能把入口线程池拖垮。方法队列满时的行为更适合的位置offer()马上返回失败请求线程提交任务put()阻塞等待空间后台生产者且允许等待add()队列满时抛异常不适合作为过载分支在任务系统中队列满不是代码异常而是一个明确的过载信号。5、Worker 如何从队列中取任务并执行任务进入队列之后需要有后台线程不断从队列中取任务。这个角色就是 Worker。Worker 的基本职责很简单等待任务、取出任务、交给 Processor 执行然后继续等待下一个任务。publicclassTaskWorkerimplementsRunnable{privatefinalBlockingQueueTasktaskQueue;privatefinalTaskProcessorprocessor;publicTaskWorker(BlockingQueueTasktaskQueue,TaskProcessorprocessor){this.taskQueuetaskQueue;this.processorprocessor;}Overridepublicvoidrun(){while(true){try{TasktasktaskQueue.take();processor.process(task);}catch(InterruptedExceptione){Thread.currentThread().interrupt();break;}catch(Exceptione){e.printStackTrace();}}}}这里 Worker 使用的是take()。它和提交端的offer()正好形成对应关系方法使用方队列满或空时的行为offer(task)提交线程队列满了马上返回失败take()Worker 线程队列空了阻塞等待Worker 可以用take()是因为 Worker 本来就是后台工作线程它的职责就是等待任务。队列为空时让 Worker 阻塞等待比让它不停空转更合理。不推荐在死循环里不断poll()因为队列为空时它会马上返回null导致 Worker 空转消耗 CPU。具体业务逻辑放在TaskProcessor中publicclassTaskProcessor{publicvoidprocess(Tasktask){task.markRunning();if(task.getType()TaskType.FILE_PARSE){parseFile(task.getBizId());generatePreview(task.getBizId());notifyUser(task.getBizId());task.markSuccess();return;}thrownewIllegalArgumentException(Unsupported task type);}privatevoidparseFile(StringfileId){// parse file}privatevoidgeneratePreview(StringfileId){// generate preview}privatevoidnotifyUser(StringfileId){// notify user}}这样 Worker 和 Processor 的职责是分开的对象关心的问题Worker如何从队列取任务如何循环执行Processor某类任务具体怎么处理Task当前任务是谁、是什么状态启动多个 Worker 时只需要按workerCount创建多个TaskWorker线程并统一管理它们的启动和关闭即可。至此单机任务系统的最小执行闭环已经形成创建任务放入队列Worker 从队列取出任务Processor 执行业务逻辑最后更新任务状态。6、Worker 数量和队列容量如何设计Worker 数量决定系统同时执行多少个任务队列容量决定系统最多允许多少任务等待。二者必须一起看不能只调其中一个。如果 Worker 太少处理速度跟不上提交速度队列会持续堆积如果 Worker 太多也不一定更快。文件解析任务可能会读文件、写数据库、调用通知接口。Worker 太多时任务虽然并发执行了但数据库连接池、磁盘 I/O、HTTP 下游可能被打满单个任务耗时反而变长。所以 Worker 数量的上限不只由线程决定还受任务类型和下游资源影响任务类型主要瓶颈Worker 数量倾向CPU 密集型CPU 计算接近 CPU 核数I/O 密集型数据库、文件、HTTP可以大于 CPU 核数但受下游限制混合型CPU I/O从保守值开始压测调整文件解析任务通常是混合型任务。它既可能消耗 CPU也可能访问文件系统和数据库。因此第一版不要开得过大可以先从保守值开始intworkerCount8;BlockingQueueTaskqueuenewArrayBlockingQueue(500);TaskProcessorprocessornewTaskProcessor();TaskWorkerPoolworkerPoolnewTaskWorkerPool(queue,processor,workerCount);workerPool.start();队列容量也不是越大越好。队列越大系统越能吸收短时流量波动但任务等待时间也会变长。队列太大还会隐藏过载让系统很晚才暴露问题。队列容量更合理的估算方式是从业务可接受等待时间倒推queue capacity ≈ process speed × acceptable wait time process speed ≈ worker count / average task cost假设有 8 个 Worker单个文件解析任务平均耗时 200ms那么系统大约每秒处理8 / 0.2 40 tasks / second如果业务允许任务最多等待 10 秒队列容量可以先估算为40 × 10 400这个数字不是最终答案但它比随手写10000更有意义。因为它表达的是系统最多缓存大约 10 秒的任务量。Worker 数量和队列容量最终都要通过实际指标校准重点观察队列长度、任务等待时间、任务执行时间、CPU 使用率、数据库连接池使用率、失败率和超时率。任务系统不能只靠经验参数必须配合监控一起设计。7、队列满了以后系统如何自保有界队列一定会出现满队列的情况。队列满不是 bug而是系统告诉调用方当前处理能力已经达到边界。如果taskQueue.offer(task)返回false系统必须做选择。这个选择就是拒绝策略。最直接的策略是快速拒绝队列满了就返回REJECTED。这种策略适合用户可以稍后重试或者任务时效性很强的场景。比如某些通知任务已经严重堆积再继续接收只会让用户更晚收到通知甚至失去业务意义。另一种策略是调用方自己执行也就是反压队列满了就让提交线程直接调用processor.process(task)然后返回CALLER_RUN。它的效果是提交线程被迫变慢从源头减轻队列压力。不过这个策略不能乱用。如果任务很慢而调用方是请求线程那么请求线程会被拖住入口线程池可能也被影响。它更适合任务较短、调用方允许变慢、并且任务不希望轻易丢弃的场景。不同任务不应该使用同一种过载策略任务类型队列满后的策略文件解析可以拒绝让用户稍后重试用户通知可以降级或延后日志上报可以采样或丢弃核心状态同步不适合只放内存队列这里先不展开“不能丢”的任务如何处理因为那已经超出单机内存队列的边界。只需要先记住队列容量定义了系统的缓冲边界拒绝策略定义了越过边界后的自保方式。8、任务失败后如何重试和隔离任务被 Worker 取出后也可能执行失败。比如文件不存在、文件格式错误、数据库写入失败、通知接口超时等。如果 Worker 只是打印异常try{TasktasktaskQueue.take();processor.process(task);}catch(Exceptione){e.printStackTrace();}那么任务失败以后就没有后续处理。系统既不知道任务是否要重试也不知道失败原因是否需要保留。更合理的做法是把失败处理从 Worker 中拆出来publicclassTaskWorkerimplementsRunnable{privatefinalBlockingQueueTasktaskQueue;privatefinalTaskProcessorprocessor;privatefinalTaskFailureHandlerfailureHandler;Overridepublicvoidrun(){while(true){Tasktasknull;try{tasktaskQueue.take();processor.process(task);}catch(InterruptedExceptione){Thread.currentThread().interrupt();break;}catch(Exceptione){if(task!null){failureHandler.handle(task,e);}}}}}Worker 仍然只负责取任务和调用 Processor。任务失败后是否重试、是否进入死信由TaskFailureHandler决定。一个简单的重试策略可以先只控制最大重试次数例如task.getRetryCount() maxRetryCount。失败处理器可以这样写publicclassTaskFailureHandler{privatefinalBlockingQueueTasktaskQueue;privatefinalRetryPolicyretryPolicy;publicvoidhandle(Tasktask,Exceptione){Stringerrore.getMessage();task.markFailed(error);if(shouldRetry(e)retryPolicy.canRetry(task)){task.increaseRetryCount();if(!taskQueue.offer(task)){task.markDead(queue full when retry: error);saveDeadTask(task);}return;}task.markDead(error);saveDeadTask(task);}privatebooleanshouldRetry(Exceptione){returneinstanceofTimeoutException||einstanceofIOException;}privatevoidsaveDeadTask(Tasktask){// save to log, local file, monitoring system, or later persistent storage}}这段逻辑表达的是重试必须有边界。不能因为任务失败就无限重试否则失败任务会一直占用 Worker。尤其当下游服务已经异常时无限重试会持续放大故障。也不是所有异常都适合重试失败类型是否适合重试例子临时失败适合网络抖动、HTTP 超时资源繁忙适合延迟重试下游限流、连接池满参数错误不适合文件 ID 不存在业务规则失败不适合当前状态不允许处理代码 Bug不适合盲目重试空指针、类型转换错误如果失败是临时性的最好不要立刻重试。因为马上重试很可能再次失败。更合理的是延迟重试比如第一次失败后 5 秒重试第二次失败后 30 秒重试第三次失败后 2 分钟重试。本章不展开DelayQueue的实现只需要先建立设计原则重试用于恢复临时失败死信用于隔离长期失败。重试不能无边界死信也不是简单丢弃而是为后续排查和补偿保留入口。9、任务系统如何监控和优雅关闭任务系统跑起来以后还需要知道它是否健康。不能只看服务进程是否存活因为任务系统最常见的问题不是立刻宕机而是队列慢慢堆积、任务越等越久、失败率逐渐升高。最核心的指标包括队列长度、队列容量使用率、任务等待时间、任务执行时间、成功数、失败数、重试次数、死信数量和 Worker 存活数。为了统计等待时间和执行时间可以给Task增加时间字段privatefinallongcreateTime;privatelongstartTime;privatelongfinishTime;publicvoidmarkRunning(){this.statusTaskStatus.RUNNING;this.startTimeSystem.currentTimeMillis();}publicvoidmarkSuccess(){this.statusTaskStatus.SUCCESS;this.finishTimeSystem.currentTimeMillis();}publiclonggetWaitTime(){returnstartTime-createTime;}publiclonggetExecuteTime(){returnfinishTime-startTime;}这两个指标能帮助定位问题如果等待时间变长但执行时间正常说明 Worker 处理能力可能不足或者任务提交速度突然升高。如果执行时间变长说明业务逻辑或下游资源变慢Worker 被慢任务占住后续等待时间也会跟着升高。队列指标可以定期采集intqueueSizetaskQueue.size();intremainingCapacitytaskQueue.remainingCapacity();intcapacityqueueSizeremainingCapacity;doubleusagequeueSize*1.0/capacity;除了运行时监控还要考虑服务关闭。关闭时 Worker 可能正在等待任务也可能正在执行任务队列里也可能还有任务。如果直接停进程内存队列里的任务会丢失正在执行的任务也可能只执行一半。单机内存任务系统的优雅关闭至少要做到三步可以给 WorkerPool 增加运行状态publicclassTaskWorkerPool{privatevolatilebooleanrunningtrue;publicbooleanisRunning(){returnrunning;}publicvoidshutdown(){runningfalse;for(Threadworker:workers){worker.interrupt();}}}Worker 循环可以从永久take()改成带超时的poll()这样它能周期性检查关闭状态publicvoidrun(){while(pool.isRunning()||!taskQueue.isEmpty()){try{TasktasktaskQueue.poll(1,TimeUnit.SECONDS);if(tasknull){continue;}processor.process(task);}catch(InterruptedExceptione){Thread.currentThread().interrupt();break;}catch(Exceptione){// handle failure}}}关键在循环条件while(pool.isRunning()||!taskQueue.isEmpty())它的含义是只要系统还在运行就继续取任务即使系统已经开始关闭只要队列里还有任务也尽量继续处理只有系统已经关闭并且队列也空了Worker 才退出。关闭后还需要等待 Worker 结束但不能无限等。通常会用join(timeout)或线程池的awaitTermination给 Worker 一个有限的完成窗口。这样关闭流程就有了边界不过要注意内存队列只能做到“尽量处理”。如果机器宕机、进程被强杀内存里的任务仍然会丢失。正在执行的任务是否能恢复也没有可靠记录。这正是单机内存任务系统的边界它可以解决并发执行和短暂缓冲但不能保证任务在重启、宕机、多实例部署下仍然可靠。总结并发任务处理系统的起点是请求线程不适合直接承担所有耗时工作。为了让请求处理和后台执行解耦系统需要把业务动作抽象成Task再通过有界队列把任务提交速度和任务执行速度隔开。有了任务和队列之后Worker 负责从队列中取任务Processor 负责执行具体业务逻辑。Worker 数量决定并发执行能力队列容量决定短时缓冲能力。二者都不能无限放大因为真正的限制往往来自 CPU、数据库、文件系统、HTTP 下游和业务可接受延迟。当队列满了系统必须通过拒绝、降级、调用方执行等策略保护自己当任务失败了系统必须通过有限重试和死信隔离让失败能够收敛当系统运行起来以后还需要通过队列长度、等待时间、执行时间、失败率和 Worker 存活数判断它是否健康当服务关闭时也要先停止接收新任务再尽量处理完已进入队列的任务。因此单机内存任务系统解决的是“有限资源下如何有边界地并发处理任务”。它适合轻量异步、短暂削峰和允许一定丢失风险的任务。但如果任务不能丢需要跨机器共享或者服务重启后还要恢复执行就必须继续引入数据库任务表或 MQ。那已经是可靠任务系统要解决的问题。