从并发运行到并行管理：批量任务调度系统的状态机与资源保护实践

去年做商品数据批处理一次批量拉取三千多个SKU的数据我图省事直接开了50个线程去跑。前两分钟一切正常第三分钟数据库连接池被打满线上核心接口开始超时最后整个模块被网关熔断领导在群里问了一句“谁在跑批量任务”我到现在都记得那个感觉。复盘的时候我才意识到一个问题我能写并发的代码不代表我能管好并行的任务。“并发运行”解决的是任务怎么跑起来的问题而“并行管理”解决的是让谁跑、跑几个、跑到一半挂了怎么办的问题。后者才是真正决定系统稳不稳定的关键。这篇文章我想把这段时间沉淀下来的东西完整梳理一遍围绕“并行管理与并发运行”这个话题讲清楚任务状态机怎么设计、并发上限怎么定、限流怎么配、任务编排怎么做、可观测性怎么搭最后给一个可以直接抄的Demo实现。适合刚开始接触批量任务调度、异步队列、多线程管理的开发者也适合那些已经把系统跑起来但总是不放心的工程师。1. 一次线上事故让我重新理解了“并行”和“并发”先还原一下事故现场。当时我们的商品服务需要从多个第三方渠道同步价格和库存每个SKU对应一到三个请求。我设计了一个同步模块用Python的ThreadPoolExecutor线程数直接设成50任务提交完就不管了。表面上看一切合理50个线程确实把同步速度提升了但问题出在“不管了”这三个字上。50个线程同时执行时每个线程都要拿数据库连接来查询SKU的基础信息一个连接池总共才20个连接。50个请求同时涌进来20个连接不够用剩下的30个线程全部阻塞等待。业务接口本来也要依赖这个连接池现在被批量任务占满接口响应时间从50ms直接跳到30秒最后网关熔断。事故的原因不是“并发”本身而是我没有管好“并行”的两个维度并发运行任务确实多线程跑了这是执行层面的。并行管理没有限制并发上限、没有保护下游资源、没有优先级区分这是治理层面的。很多刚接触多线程的开发者觉得“并行管理”就是锁、原子操作、线程安全但真正到了任务级批量处理的场景管理的是任务的完整生命周期入队、调度、执行、超时、失败、重试、取消、回收。我后来查了一些团队自研的调度系统发现它们做的事情本质上也是一样的只是把这张表从内存搬到了数据库。那次事故后我把这个模块重写了一遍核心改动只有四个线程数从50压到10配合信号量做双重控制。所有任务进入队列先排队再执行。数据库连接池单独配置不跟业务接口抢。每个任务记录生命周期失败可以查、可以重试。这套改动上线后同步速度虽然比50线程时候慢了一点——从20分钟变成35分钟但核心接口的P99稳定在60ms以内再也没出过事故。这个经历让我明确了一点并发运行是“术”并行管理是“道”。当任务的规模从几十涨到几千、几万时决定系统上限的一定不是线程池怎么写而是任务有没有状态、状态有没有被正确流转、异常有没有被捕获和处理。接下来我就按这个思路展开。2. 任务状态机并行系统能不能立住的根基我接触过的很多并行系统问题都出在同一个地方任务没有明确的状态。任务提交到线程池之后执行成功就在日志里打一行失败就抛个异常超时就挂在那里。整个系统像是一个没有红绿灯的路口看起来车在动但没人知道每辆车要去哪、走完没有。2.1 状态定义与字段设计一个合格的并行任务系统任务至少要包含这些状态状态字段值含义触发时机排队中PENDING已提交等待worker消费任务入队运行中RUNNINGworker已取出正在执行worker开始成功SUCCESS执行完成结果校验通过无异常返回失败FAILED执行抛出异常重试耗尽异常且不再重试超时TIMEOUT超过设定时间未完成看门狗检测取消CANCELED被用户或上层系统中止主动取消很多人觉得这个表太简单了但真正落到代码里你会发现“状态”不只是内存里的一个枚举而是一组被读写和流转的数据。我在实际的代码中会给Task加这些字段class Task: def __init__(self, task_id, func, timeout10): self.task_id task_id # 全局唯一ID self.status PENDING self.retry_count 0 # 已重试次数 self.max_retry 3 # 最大重试次数 self.timeout timeout # 超时时间秒 self.started_at None # 启动时间用于算耗时 self.finished_at None # 结束时间 self.trace_id None # 日志追踪ID self.result None # 执行结果 self.error_msg None # 异常信息这些字段看着琐碎但跑起来之后全是救命的东西。started_at和finished_at算耗时trace_id排查问题retry_count控制重试error_msg直接透出到监控面板。2.2 状态流转里的三个反直觉细节状态定义不难难的是状态怎么“流转”。我踩过三个坑全部都是反直觉的第一个坑运行中的任务能不能取消很多人说能——用Future.cancel()不就行了。但Java的Future.cancel()只对还没开始执行的任务有效一旦线程已经拿着任务在跑cancel()只是打个中断标记如果你的任务不响应中断取消就是一句空话。在Python里面甚至更直接Thread没有官方的中断机制。所以我的方案是取消操作只改状态标识真正的中断交给应用层自己处理。任务执行过程中定期检查自己的状态发现被标记为CANCELED就主动退出。系统层面的“强制取消”只对进程级别的运行有效线程级别的强杀永远是不安全的。第二个坑失败任务要不要直接进终态第一次设计状态机时FAILED我直接当终态处理。结果第二天看数据一批网络抖动导致的任务全部FAILED日志里白花花一片异常还要手动写脚本去补偿。后来我把状态流改成了这样任务第一次抛异常时不直接标记FAILED而是重新入队把retry_count加一。只有重试次数达到上限时才从RUNNING变成FAILED。失败不应该是一个瞬时结论而应该是一个经过重试确认后的最终判定。第三个坑超时检测不要依赖任务自己报告。这个坑最隐蔽。刚开始我让任务自己在try/finally里面处理超时但现实是任务卡死了finally根本执行不到线程永远被占据任务的状态永远停在RUNNING。正确的做法是由独立的看门狗线程去扫描正在运行的任务比对started_at timeout和当前时间超时就强制标记为TIMEOUT并释放线程资源。2.3 存储层的状态同步乐观锁别省略任务状态一旦落到数据库并发问题就来了。多个worker同时在跑每个完成之后都要更新自己的状态。如果不加锁两个线程可能同时读到PENDING同时更新后写的覆盖先写的状态就乱了。我最开始用的是“先查再改”的方式后来线上出现了任务状态从SUCCESS被改回RUNNING的怪事排查半天就是并发更新导致的覆盖。解决方案很简单在状态表里加一个version字段更新时带上WHERE version ?每次更新成功把version加一。如果更新影响行数为0说明版本冲突重新查任务状态再做决策。这是最朴素的乐观锁思路但对付这类场景绰绰有余。3. 并发运行的三根保险丝并发上限、限流、资源保护我把上一节讲到的状态机当成并行系统的骨架那这一节的内容就是系统的保险丝。没有保险丝电压一高整个电路就烧了没有并发控制任务一多整个服务就挂了。3.1 并发上限不是拍脑袋定的回到开头那次事故我把线程数设为50就是一个典型的“拍脑袋”。后来我总结出一套还算靠谱的估算思路先看下游能扛多少再定并发上限。举个例子我们的同步任务需要调用第三方商品API这个API的QPS上限是100。单次请求的平均耗时是200ms。那么单线程一秒钟最多处理5个请求想要不触发下游限流并发线程数最多就是100 / 5 20。这是理论值实际还要留30%的余量所以我最终设为14。公式可以总结成并发上限 下游QPS上限 × 单请求耗时秒 × 安全系数安全系数一般取0.6到0.8。如果你的下游是数据库连接池道理一样连接池有20个连接那么任何时刻执行任务时并发持有连接的数量就不能超过20否则就有线程被阻塞的风险。代码层面我用的是“线程池固定大小 信号量双重控制”import threading from concurrent.futures import ThreadPoolExecutor # 线程池大小控制worker数量 executor ThreadPoolExecutor(max_workers10) # 信号量控制并发访问下游的数量 semaphore threading.Semaphore(14) def guarded_task(task): with semaphore: task.execute() for task in tasks: executor.submit(guarded_task, task)这样做的意义是线程池管“这个系统最多同时跑多少个任务”信号量管“这些任务里有多少能被允许同时访问下游”。两者解耦调整任意一侧都更方便。3.2 限流保护的永远是对端很多人把“限流”理解成保护自己的服务不被压垮但在并行任务场景里限流保护的是下游服务。尤其是调用第三方API时人家的限流策略是固定窗口还是令牌桶你不知道一旦触发限流返回的是一堆429或者503重试反而加重了对方的压力。我在项目里做了两层防护应用层限流用令牌桶算法自身令牌生成速率对齐下游的QPS并对任务做重试退避。熔断层保护如果下游连续返回错误比例超过阈值直接打开熔断开关不再发起新请求等冷却期结束再半开试探。Google Guava的RateLimiter做单机令牌桶很方便但注意它是单JVM维度的限制。如果服务部署了多份限流总量就不是单台机器的值而是所有实例之和。这时候要么用Redis做分布式限流要么把并发上限除以实例数别想当然。3.3 共享资源保护连接池、锁与幂等并行任务系统里最常见的共享资源是数据库连接、文件句柄、外部接口令牌。保护手段分两种连接池管理给批量任务单独配置一个连接池不要和业务接口共用。业务接口对延迟敏感批量任务对吞吐量敏感两者混用一定会互相伤害。分布式锁多机部署时两个机器可能同时处理同一个任务。比如定时任务轮询数据库时任务状态是PENDING两台机器同时捞出来就会重复执行。解决方案是SELECT ... FOR UPDATE SKIP LOCKED跳过已经被锁的行或者用Redis的SETNX做分布式锁。有一个观点我特别认同在并行系统里幂等比事务锁更重要。锁只能解决“同一时间只有一个执行者”的问题但解决不了“执行者挂了然后重试结果执行了两次”的问题。所以每个任务必须带上业务幂等键处理端做去重。比如商品同步任务可以用”sku_id channel sync_date“作为幂等键落到数据库里加唯一索引。重复请求来了要么被唯一索引挡住要么直接覆盖写同一行结果一致。4. 从“单机跑通”到“批量编排”调度器的进化路径状态机和并发控制搭好了系统已经能稳定跑一批任务了。但如果你管理的是几百上千个任务会发现还需要一层更上层的东西调度器。调度器不是线程池它解决的是“任务之间的关系”问题。4.1 批次管理给任务带上batchId我在做SKU同步的时候一次操作可能提交3000个任务。如果把它们当成3000个独立任务监控和排查都是灾难——出问题时你不知道是哪个批次的锅。所以我引入了batchId的概念。一批任务共享同一个批次号批次本身也有状态RUNNING、PARTIAL_SUCCESS部分成功、ALL_SUCCESS、FAILED。任务维度的状态机解决“单个任务是否健康”批次维度的状态解决“一次需求的执行情况是否满足预期”。新增一个批次的业务含义还带来了一个好处同一批次的任务可以用相同的并发策略。比如某个批次依赖的下游QPS比较低就把这个批次的信号量设小一点下游扩容了就调大。不同批次之间互不影响做到了并发策略的隔离。4.2 失败重试指数退避与重试上限我见过最朴素的失败重试写法是这样的for i in range(3): try: do_something() break except: time.sleep(1)这种代码在小规模任务里问题不大但放到并行任务系统里如果3000个任务同时失败同时sleep 1秒再重试下游会收到3000个请求的同时重试。实际上这就是一种“重试风暴”。正确的做法是指数退避加上随机抖动import random import time def retry_with_backoff(func, max_retry3): for attempt in range(max_restry): try: return func() except Exception as e: if attempt max_retry - 1: raise wait_time (2 ** attempt) random.uniform(0, 0.5) time.sleep(wait_time)指数退避让重试的音量在时间轴上摊开随机抖动是防止多个任务在同一毫秒醒来。这两件事合起来才算是把“并发重试”这件事做得合格。另外要区分可重试错误和不可重试错误。网络超时、连接重置、503、429这些是可重试的参数错误、业务规则校验失败、401鉴权失败这些是不可重试的。可重试的才放进重试逻辑不可重试的直接标记FAILED。把不可重试的任务反复跑三遍浪费的是所有排队任务的等待时间。4.3 最简DAG依赖编排的落地方式有些任务之间有依赖关系。比如我先要拉取商品详情再根据详情中的图片链接去拉取图片。如果不做依赖管理两种做法最常见串行执行先跑第一层全部结束再跑第二层简单但慢。每层手动提交写一堆回调代码越写越绕。我推荐一个最简的DAG实现思路在任务里加一个depends_on字段记录上游任务ID列表调度器在派发任务之前先检查依赖是否全部完成如果没完成就放回队列等下次调度。伪代码如下def is_ready(task): for dep_id in task.depends_on: if task_map[dep_id].status ! SUCCESS: return False return True这套方案在任务量几千级别完全够用逻辑清晰、排错容易。如果你的任务关系复杂到几百个节点互相依赖建议直接上现成的DAG工作流引擎比如Airflow或者Temporal别自己硬造轮子。4.4 从单机到多机扩容后的两个坑单机跑得稳不代表多机跑得稳。我把批量同步模块从单机部署扩展到三台机器后遇到了两个典型的坑坑一任务被重复领取。单机方案里任务是线程安全的一个任务只会被一个线程拿到。多机之后多个进程同时扫数据库里的PENDING任务就会出现两个进程领到同一个任务。上面提到的SELECT ... FOR UPDATE SKIP LOCKED就是解决方案确保同一时刻只有一个worker能锁住某一行。坑二任务状态的分布式可见性。原来状态存在内存里一个进程内部没问题。多机之后A机器的任务成功了B机器不知道可能导致依赖它的后续任务被阻塞。所以多机场景下任务状态必须落到共享存储Redis或数据库依赖检查都从共享存储读。5. 可观测性并行系统不能是黑盒并行系统的痛点是症状和原因之间的时间间距可能很长。某个任务失败可能是五分钟前下游抖动导致的也可能是昨天的代码引入的bug。如果没有可观测性排查问题就像在黑盒子里抓瞎。5.1 用trace_id把任务链路串起来我强烈建议每个任务在提交时生成一个trace_id然后让这个ID贯穿任务的全生命周期任务表里记录它日志里打印它调用外部API的HTTP头里带上它最终监控面板上展示它。举例说明作用。一次线上任务大量失败我随便抽了一条日志[trace_id: sku_sync_20240801_001234] request failed, status500, responsetimeout然后我用这个trace_id去数据库里查SELECT status, retry_count, error_msg, started_at, finished_at FROM task WHERE trace_id sku_sync_20240801_001234;一秒不到就定位到这个任务调用的第三方API在某个时间段内大面积超时错误信息是timeout重试三次后进入FAILED状态。如果没有trace_id我在几千条日志里搜关键字运气好能搜到运气差要翻半天。5.2 状态回收与孤儿任务清扫不管状态机设计得多好总会存在一些“僵死”的任务。比如worker进程被kill -9杀掉任务状态永远停在RUNNING。数据库临时抖动更新状态失败任务状态停留在PENDING但实际已经执行过了。下游响应延迟巨大任务被看门狗标记为TIMEOUT但线程还没释放。这些问题单靠执行流程是规避不了的需要一套定时清扫机制。我每个小时跑一次清扫任务找出所有状态为RUNNING但started_at超过当前时间两倍于任务超时阈值的记录强制标记为FAILED并写入error_msg worker crashed or task hung。找出所有状态为PENDING但创建时间超过24小时的记录标记为CANCELED。找出重试次数已经大于等于max_retry但仍停留在RUNNING的异常记录强制终结。清扫脚本本身要单独部署不要放在同一个worker进程里否则你杀的就是你自己家的环卫工。5.3 四个必须盯的监控指标我用的监控面板核心就四个指标多了也看不过来指标计算方式告警规则并发运行中任务数状态为RUNNING的任务计数大于上限持续5分钟告警队列深度状态为PENDING的任务计数大于500且持续10分钟告警任务成功率SUCCESS / (SUCCESS FAILED)低于95%告警P99耗时按batchId分组的耗时百分位比历史均值高2倍告警这四个指标组合起来能回答“要不要扩容”“下游是不是挂了”“当前的任务是否积压”这三个高频问题。我个人经验是先盯队列深度和成功率这两个指标出的问题覆盖了80%的线上事故。6. 一个能直接跑的任务管理器Demo前面讲了那么多如果不上手跑一遍理解总是隔着一层。我写了一个用Python标准库实现的最小任务管理器不依赖任何第三方包能跑、能看状态、能超时检测、能统计结果。代码逻辑和前面讲的设计一一对应可以作为你业务代码的起点。6.1 Demo的边界与设计这个Demo只做四件事固定数量的worker并发消费任务队列。任务有PENDING/RUNNING/SUCCESS/FAILED/TIMEOUT五种状态。看门狗线程检测超时任务。主线程等待所有任务结束。不做的有数据库持久化、失败重试、批量编排、分布式部署。这些前面已经讲清楚了思路根据业务需求自行扩展。6.2 核心代码import queue import threading import time import traceback from enum import Enum class TaskStatus(Enum): PENDING pending RUNNING running SUCCESS success FAILED failed TIMEOUT timeout class Task: def __init__(self, task_id, func, args(), kwargsNone, timeout3.0): self.task_id task_id self.func func self.args args self.kwargs kwargs or {} self.timeout timeout self.status TaskStatus.PENDING self.result None self.error None self.started_at None self.finished_at None class TaskRunner: def __init__(self, max_workers4): self.max_workers max_workers self.task_queue queue.Queue() self.running {} self.lock threading.Lock() self.workers [] self.watchdog None def start(self): for i in range(self.max_workers): t threading.Thread(targetself._worker_loop, namefworker-{i}, daemonTrue) t.start() self.workers.append(t) self.watchdog threading.Thread(targetself._watchdog_loop, namewatchdog, daemonTrue) self.watchdog.start() def submit(self, task): self.task_queue.put(task) def _worker_loop(self): while True: task self.task_queue.get() task.status TaskStatus.RUNNING task.started_at time.time() with self.lock: self.running[task.task_id] task try: task.result task.func(*task.args, **task.kwargs) task.status TaskStatus.SUCCESS except Exception: task.error traceback.format_exc() task.status TaskStatus.FAILED finally: task.finished_at time.time() with self.lock: self.running.pop(task.task_id, None) self.task_queue.task_done() def _watchdog_loop(self): while True: time.sleep(0.5) now time.time() with self.lock: for task in list(self.running.values()): if now - task.started_at task.timeout: task.status TaskStatus.TIMEOUT task.error ftask {task.task_id} timed out after {task.timeout}s self.running.pop(task.task_id, None) def wait(self): self.task_queue.join() def demo_task(task_id, sleep_time): time.sleep(sleep_time) return ftask_{task_id}_done if __name__ __main__: runner TaskRunner(max_workers4) runner.start() for i in range(12): timeout 2.0 if i % 3 0 else 5.0 sleep_time 1.0 if i % 3 0 else 4.0 runner.submit(Task(task_idi, funcdemo_task, args(i, sleep_time), timeouttimeout)) runner.wait() for task_id in range(12): task runner.running.get(task_id)这个版本有个取舍为了代码简短没有维护全局的task_map来记录所有任务结果你可以在wait()之后自行维护一个字典。看门狗检测到超时后只是改状态没法杀线程——前面说过线程强杀在标准库里做不到生产环境建议用进程隔离或者配合进程池来实现强制终止。6.3 运行效果与分析跑这个Demo你会看到4个worker开始几乎同时取出前4个任务状态从PENDING变成RUNNING。随后每完成一个任务worker从队列里拿下一个。4个worker的并行效果显著12个任务每个平均sleep 3秒左右串行要36秒并行执行首次全部结束大概在12秒左右吞吐量提升约3倍。有意思的是那些timeout2.0的任务它们sleep了1秒能在2秒内完成所以不会超时。如果你把sleep时间改成3秒看门狗就会在2秒时把它们标记为TIMEOUT。试试这个调整你能直观感受到看门狗线程的作用。6.4 换语言怎么迁移这套设计这套“任务队列 固定worker 状态机 看门狗”的设计是语言无关的。我在其他语言里也实现过类似的东西Go用channel替代队列goroutine替代线程context.WithTimeout实现超时控制天然支持高并发。Java用ExecutorServiceFutureTaskCompletionService管理已完成任务看门狗用ScheduledExecutorService定时扫描。C#TaskChannelCancellationTokenSource和Go的思想非常接近。换语言时只要抓住三个核心概念——队列、worker池、状态存储——就不会跑偏。7. 踩坑清单与我的个人体会最后一部分不说空话直接把我在这类系统上踩过的坑整理成清单你可以拿来做对照检查。7.1 六条用真金白银换来的经验1. 线程数不是越大越好越大越容易把自己打垮这是我事故的直接原因。线程数的上限不由“CPU核心数”决定而由“下游能扛多少”决定。先评估下游的QPS上限和单请求耗时再倒推并发上限最后乘上安全系数。2. 任务失败一定要重试但不带退避的重试是耍流氓没有退避的并发重试会造成重试风暴把下游从“偶发失败”打成“持续不可用”。指数退避加随机抖动是底线别用固定sleep。3. 超时检测必须独立于任务执行线程别让任务自己负责检测超时卡死的任务永远不会报告自己卡死了。独立的看门狗线程扫描运行中任务是更可靠的方案。4. 任务状态更新必须走乐观锁不加version字段并发更新就会互相覆盖。这个坑只在特定条件下出现条件不满足时你是测不出来的等线上数据出问题才惊讶为什么状态回退了。5. 线上出问题没有trace_id就是灾难没有trace_id排查并行任务问题就像在一堆火柴里找一根烧过的理论上能找到但时间成本太高。任务ID无论如何都要生成。6. 并发测试只能在生产环境测本地测不出来本地机器的网络、数据库、第三方服务都和一个生产环境的真实负载差着数量级。凡是涉及并发的参数本地调通了不等于生产环境能扛住。上线前做一次小流量的生产验证比写一百个单元测试都有用。7.2 这套设计的适用边界与扩展方向讲实话如果你只有几十个任务并发执行把状态机做这么重可能有点过度设计。但任务的规模过千之后没有状态管理、没有重试、没有监控任何一个环节出问题都够你喝一壶。我自己的判断标准是当任务的失败不再是个例而是成批出现时就说明你需要从“并发运行”升级到“并行管理”了。这套设计目前适合中小规模的批量任务系统几百到几千并发任务都是没问题的。如果任务规模到数万、执行链路更长、需求更复杂就别自己造轮子了直接选一个成熟的工作流引擎更划算。学习这套东西的价值在于你用了别人家的系统也能看懂它的状态表、重试策略和调度逻辑出问题时不至于两眼一抹黑。