1. ax调度到底是什么它解决了什么问题1.1 从一个真实的业务痛点说起三年前我们团队接手了一套商城系统当时线上经常出现一种诡异的现象用户下单后优惠券到账要延迟三五分钟账单日批量跑批任务经常卡死半夜的定时报表偶尔就漏掉一份。每次排查代码逻辑都没问题纯粹是任务调度这块太乱——有人用Linux crontab硬跑有人把定时逻辑写在业务代码里轮询还有人直接用Thread.sleep加死循环。后来我们把所有异步执行的场景收拢到一起做了一套内部代号叫ax的调度引擎。全称是Async eXecution Scheduler大家叫顺口了就说“ax调度”。这套东西把三类问题一次性解决了定时任务能不能准时触发异步任务能不能被可靠地执行执行失败之后能不能自动恢复。简单说它管的是“任务到底什么时候跑、交给谁跑、跑挂了怎么办”。1.2 ax调度的核心定位与使用场景ax不是业务流程本身它是业务和底层资源之间的一层“时间与可靠性代理”。你可以把它理解为餐厅里的叫号系统顾客业务任务来了先取号系统调度引擎负责任务排序和通知叫到号之后后厨执行器才开始做菜。没有这套叫号机制所有顾客一窝蜂挤到后厨窗口谁先谁后全靠嗓门结果必然是乱套。它最典型的应用场景包括电商场景下单后延迟关闭未支付订单、支付成功后的异步通知、积分/优惠券发放数据场景每天凌晨的报表聚合、小时级的缓存预热、数据同步任务运维场景定时健康检查、日志归档、容量巡检业务解耦场景大流量瞬间产生的削峰填谷比如秒杀后的异步订单处理如果你团队里已经有超过三种不同的定时任务实现方式或者线上出现过“这次任务到底跑没跑”的争论那基本就到了需要一套统一调度引擎的临界点。1.3 适用人群与前置知识这篇文章写给谁主要是有一定后端开发经验、但还没自研过调度系统的工程师或者是小团队的技术负责人业务量还没大到必须上XXL-Job、DolphinScheduler这类开源重型框架但又不想再用crontab糊弄事了。你最好对消息队列的基本概念不陌生知道Redis的Stream结构懂一点Python或Go的并发模型就能跟完整个实操过程。我不打算把文章写成源码逐行解析而是从“如果让你从零搓一个ax调度你会怎么设计”的角度把架构选型、核心代码、参数调优、线上踩坑全部捋一遍。你不需要照着抄但你一定能从里面找到自己做调度系统时该避开的坑。2. 整体设计一个调度引擎最核心的模型拆解2.1 三类任务模型定时、延迟、周期动手写代码之前最先要定的是任务模型。我见过不少团队上来就写一个execute(task_id)的接口把所有逻辑混在里面最后调度器的代码变成一堆if-else。ax的做法是把任务按触发性质拆成三大类每一类对应不同的时间计算策略任务类型触发方式典型场景时间计算策略一次性定时任务指定绝对时间点“今晚23:00执行数据库备份”直接算时间戳到点触发延迟任务相对创建时间点“下单后30分钟未支付则关单”创建时间延迟时长周期任务Cron表达式/固定间隔“每5分钟拉取一次汇率”每次完成/触发后计算下一次时间这个区分非常重要因为三种任务在存储和恢复逻辑上完全不同。一次性定时任务如果节点重启只要任务表落库大不了重新扫描延迟任务则要特别注意时间起点是“创建时刻”而不是“调度器感知时刻”周期任务必须考虑当前周期没跑完、下一个周期已到时的重叠策略——是跳过还是并发执行必须在模型层就给出明确语义。2.2 调度器与执行器分离为什么必须拆开很多自研调度系统失败的根本原因是把“决定谁该执行”和“真正执行”混在一个进程里。这样写起来很爽部署很简单但一旦执行的任务里有一个死循环或者长休眠整个调度进程就卡住了所有任务跟着遭殃。ax的架构从一开始就坚持调度器Scheduler和执行器Worker分离。调度器是一个独立的常驻进程它只干一件事扫描任务表、计算触发时间、把到期的任务投递到队列里。真正干活的是执行器执行器是跑在业务服务里的worker模块或者是单独部署的worker集群它监听队列拿到任务描述后调用对应的业务代码。这两者之间通过一个可持久化的消息通道解耦。我选的是Redis Stream后面会详细说为什么。这个拆分带来的直接好处有两个一是调度器负载非常稳定无论业务任务多奇葩都不会拖垮触发链路二是执行器可以独立扩容大促期间直接加worker节点完全不碰调度器。2.3 一致性与幂等性任务不丢、不重的基本盘设计调度系统时有句话必须天天挂在嘴边分布式环境下任务不可能恰好只执行一次你只能在“可能重复执行”和“可能会丢”之间选边站。ax的选择是保证不丢尽量让重复变少业务侧必须做好幂等。不丢靠的是三层保障。第一层所有任务定义和触发记录都落MySQL调度器崩溃后重启重新从数据库扫描第二层到期任务投递到Redis Stream里Stream本身有持久化不会因为Redis重启就丢消息第三层执行器处理完成后必须向调度中心上报ack只有ack成功任务状态才会从RUNNING变成SUCCESS。重复的问题主要出在两类场景调度器扫表时业务长事务导致重复扫描以及执行器处理超时后被重新投递。这两个坑我在第4章会专门讲这里先记住一个原则——每个任务都必须带一个全局唯一的biz_id业务处理逻辑靠它做幂等。至于重试ax默认最多重试3次超过就进入死信队列等人工介入。3. 核心实操从零搭一个ax调度服务3.1 基础设施选型与依赖清单先把我用的是哪些组件摆出来再说为什么这么选。调度器语言Python 3.10 asyncio。选Python纯粹是团队技术栈统一换Go完全没问题核心思路一样任务元数据存储MySQL 8.0InnoDB引擎队列中间件Redis 6.2以上用Stream数据结构执行器Python进程可以是独立服务也可以嵌入业务服务可选组件Prometheus Grafana做监控关于队列选型我对比过三种方案。纯Redis List有个致命问题消费组能力弱一个消息被多个worker同时取走很难处理RabbitMQ/RocketMQ能力很强但很多团队没有专门的中间件运维光搭建和维护就是一笔成本。Redis Stream在最红的时候引入了消费者组Consumer Group支持消息ack、pending队列、消息持久化完美匹配调度器投递、worker消费、失败重投这个模型而且Redis几乎所有团队都已经在用了零额外运维成本。这套组合的代价是不能支撑海量消息但任务调度这个场景本身消息量就远小于业务消息量够用。3.2 核心表结构设计任务元数据与执行记录分离任务表的设计直接影响调度器的扫描效率这块我把踩过的坑都填进去了。ax有两张核心表ax_task_info存任务定义ax_task_instance存每一次具体的执行记录。任务定义表的核心字段CREATE TABLE ax_task_info ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_code VARCHAR(64) NOT NULL COMMENT 业务任务唯一编码, task_name VARCHAR(128) NOT NULL, task_type TINYINT NOT NULL COMMENT 1-定时 2-延迟 3-周期, cron_expr VARCHAR(64) DEFAULT NULL COMMENT 周期任务专用, delay_seconds INT DEFAULT NULL COMMENT 延迟任务专用, trigger_time DATETIME DEFAULT NULL COMMENT 定时任务触发时间, callback_url VARCHAR(256) COMMENT 执行器回调地址, status TINYINT NOT NULL DEFAULT 1 COMMENT 1-启用 0-停用, create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, KEY idx_trigger_status (status, next_trigger_time) ) ENGINEInnoDB;任务实例表记录每一次执行情况这个表是排查问题的主战场CREATE TABLE ax_task_instance ( id BIGINT PRIMARY KEY AUTO_INCREMENT, task_info_id BIGINT NOT NULL, biz_id VARCHAR(64) NOT NULL COMMENT 业务幂等ID, status TINYINT NOT NULL COMMENT 0-待执行 1-执行中 2-成功 3-失败 4-超时, retry_count INT NOT NULL DEFAULT 0, scheduler_ip VARCHAR(45) COMMENT 负责投递的调度器节点, executor_ip VARCHAR(45) COMMENT 执行的worker节点, trigger_time DATETIME NOT NULL, finish_time DATETIME DEFAULT NULL, fail_reason TEXT, UNIQUE KEY uk_biz_id (biz_id) ) ENGINEInnoDB;表结构里有几个设计细节值得展开说。第一biz_id必须建唯一索引这是幂等的最底层保障业务侧生成规则通常是“业务类型业务主键触发时间”保证同一个业务行为不会生成两个实例记录第二任务定义表索引用的是(status, next_trigger_time)这样调度器扫表时只需要WHERE status1 AND next_trigger_time NOW()这个索引让百万级任务量下的扫描也能在毫秒级完成第三任务实例和任务定义分开因为一条任务定义会生成成千上万条实例记录混在一张表里索引再优化也会被拖垮。3.3 调度主循环的核心逻辑调度器的主体是一个asyncio循环每秒钟跑一轮扫描。核心代码不长我直接贴出来注释写得更细一些import asyncio import aiomysql from datetime import datetime, timedelta SCAN_INTERVAL 1 # 扫描周期单位秒 async def scheduler_loop(pool, stream_client): while True: try: # 扫描到期任务状态启用且触发时间已到 async with pool.acquire() as conn: async with conn.cursor(aiomysql.DictCursor) as cur: await cur.execute( SELECT id, task_code, task_type, callback_url FROM ax_task_info WHERE status 1 AND next_trigger_time NOW() AND next_trigger_time NOW() - INTERVAL 30 SECOND LIMIT 500 ) due_tasks await cur.fetchall() for task in due_tasks: # 生成实例记录拿到biz_id biz_id generate_biz_id(task) instance_id await create_instance(pool, task, biz_id) # 投递到Redis Stream await stream_client.xadd( ax:task:stream, { instance_id: instance_id, task_code: task[task_code], callback_url: task[callback_url], }, ) # 计算下一次触发时间并更新任务定义表 next_time calc_next_trigger_time(task) await update_next_trigger_time(pool, task[id], next_time) except Exception as e: # 调度器本身不能崩异常全部捕获并记录 logger.exception(scheduler loop error: %s, e) await asyncio.sleep(SCAN_INTERVAL)这段代码里最容易被忽略的是next_trigger_time NOW() - INTERVAL 30 SECOND这个条件。我加它的原因是防止时钟回拨或者数据库主从延迟导致重复扫描历史数据。没有这个下限条件的话一旦某次扫描因为网络抖动多阻塞了几秒下一次扫描会把这几秒内已经处理过的任务再扫一遍产生重复投递。加上30秒的窗口意思是只处理当前时刻前后最近30秒内到期的任务超出这个范围的要么等下一轮要么已经被处理过了。另外一个细节是LIMIT 500。这个限制非常关键——调度器不能一次性扫出几十万条到期任务否则内存会暴涨而且投递过程太长会导致同一批任务触发时间严重漂移。500条一轮一秒钟一轮理论上单调度器每秒能处理500个任务触发对于绝大多数团队足够了。如果任务量真的更大就多部署几个调度器节点用task_info_id % node_count做分片但那是后话。3.4 执行器注册与任务分发机制调度器把任务投到Stream之后执行器怎么知道去哪取这就涉及到执行器的注册与分发机制。ax里执行器启动后会向调度中心可以复用调度器的HTTP接口发起注册上报自己的节点IP、端口、支持的任务类型列表。调度中心维护一个在线执行器列表投递任务时按照策略选择一个可用的执行器。我用的策略很简单按任务类型过滤出候选节点然后轮询选取。如果某个执行器连续3次心跳超时就从列表中摘除后续任务不再投给它。为什么要有“支持的任务类型列表”因为在实际业务里有些任务只能跑在特定的机器上。比如依赖内网数据库的报表任务必须投递到能访问数据库的worker节点再比如涉及某个第三方SDK的任务只有特定环境有那个SDK的授权。如果所有任务随机分发这两类任务会频繁失败。执行器侧的消费逻辑大致是这样async def worker_consumer(stream_client, executor_client): # 创建消费者组如果已经存在会报错捕获即可 try: await stream_client.xgroup_create( ax:task:stream, ax_worker_group, id0, mkstreamTrue ) except Exception: pass while True: # 每次读取一条消息阻塞最多5秒 resp await stream_client.xreadgroup( ax_worker_group, fworker-{uuid4().hex[:8]}, {ax:task:stream: }, count1, block5000, ) if not resp: continue for stream_name, entries in resp: for entry_id, fields in entries: instance_id fields[binstance_id].decode() callback_url fields[bcallback_url].decode() try: # 调用业务回调接口 result await executor_client.post(callback_url, json{ instance_id: instance_id }) if result.status_code 200: # 成功后ack消息才会从pending中移除 await stream_client.xack( ax:task:stream, ax_worker_group, entry_id ) else: # 业务失败进入重试流程稍后看重试策略 await handle_retry(instance_id, entry_id) except Exception as e: await handle_retry(instance_id, entry_id)这里有个非常容易踩的坑调用业务接口的http超时时间不能设置太长。我建议30秒为上限。很多新人把超时设成10分钟结果worker的协程全部卡在等待响应上任务队列越积越长看起来像死锁一样。如果业务本身需要跑很长时间应该由业务方异步返回一个“受理中”的凭证任务流里加中间状态而不是让调度器干等。3.5 失败重试与超时控制的完整策略任务调度的可靠性最终都要落到重试和超时策略上。ax的重试模型分三层每一层解决的故障类型都不一样。第一层是投递重试。调度器向Redis Stream写入消息后如果异常导致不确定是否写入成功会查任务实例表的状态如果实例记录还是“待执行”说明消息没投递成功补投一次如果已经是“执行中”说明投递成功但ack没回来不重复投递。这一层靠数据库状态做哨兵成本最低但要确保事务边界清晰。第二层是消费确认重试。worker拉到消息后调用业务接口失败或超时会把消息放回pending队列。Redis Stream的机制是消息被某个消费者读取但未ack时会进入Pending Entries ListPEL其他消费者不会重复读到。只有当消费者明确xack之后才会彻底移除。所以我们的处理策略是业务调用失败后不立即重新投递而是把失败信息记到日志表然后主动xack摘除这条消息让后续的“重试扫描器”按照重试次数和退避策略重新投递。这样避免了同一个消息在Stream里反复被同一个消费者读到。第三层是重试扫描器。有一个独立的定时任务每30秒扫描一次任务实例表找到状态为“执行中”但超过超时阈值系统默认30秒可在任务配置里覆盖的记录判定为超时状态改为“失败”进入重试流程。重试次数小于3次的重置状态为“待执行”更新下次触发时间为当前时间加退避间隔第一次10秒第二次30秒第三次两分钟大于等于3次的状态改为“永久失败”同时往死信表写一条记录。重试和超时是最能体现调度系统设计水平的地方很多系统“看起来能跑”但一遇到网络抖动就全线崩溃问题基本都出在这两层没做扎实。4. 生产环境关键参数与踩坑记录4.1 关键参数配置参照表这些参数不是拍脑袋定的每一个都对应线上实际的故障案例整理成表方便你对照配置参数建议值设置理由SCAN_INTERVAL1秒低于1秒会对MySQL造成不必要的扫描压力高于2秒则任务延迟可感知BATCH_LIMIT500单轮投递上限防止大批量到期任务压垮执行器消费超时时间30秒超过30秒的请求大概率有问题让worker释放出来处理其他任务业务接口读取超时30秒与消费超时匹配避免协程卡死最大重试次数3次超过3次再重试基本是浪费资源进入死信让人工处理更合理心跳上报间隔5秒兼顾实时性和开销执行器故障最多在15秒内被感知时钟回拨容忍上限30秒超过这个值的回拨直接报警需要人工介入检查NTP配置Redis Stream消息积压阈值10000条超过这个数值说明消费者跟不上了告警而不是默默堆积4.2 我踩过的三个大坑第一个坑时区引发的定时任务“阴间触发”。上线第一周就翻车了。有个任务是每天凌晨2点跑数据清洗结果连续几天凌晨4点才执行。排查了很久发现不是调度器的问题而是创建任务的运营同学在前端页面选择了“UTC时间”但后台存储统一用了业务本地时间两边差了两个小时。这个问题的根治手段是所有时间字段统一存UTC时间戳只在展示层转本地时间并且任务创建接口强制校验cron表达式含义前后端必须传一个timezone_id字段后端按服务端配置的时区解析。现在想想很基础但当时确实花了整整一个下午。第二个坑任务幂等被忽略导致重复发券。某个上线半年都没出问题的积分任务某天大促流量暴涨一个用户同时触发多笔订单同一个订单的关单通知被投递了三次用户收到了三条短信和三次积分入账。问题根源是任务定义的biz_id生成规则里用的时间戳粒度太粗同一秒钟内同一订单的不同实例生成了相同的biz_id唯一索引反而把数据写坏了。后来我把biz_id的规则改成订单号_事件类型_时间戳毫秒_4位随机数并加上一个分布式ID生成器问题才彻底解决。经验就是幂等ID的生成不能靠“看起来唯一”必须考虑极端并发下的碰撞概率。第三个坑Redis Stream的pending堆积导致重复投递。消费超时后重试逻辑设计得不严谨导致同一消息在pending里被多次xreadgroup读取。现象是任务日志里同一个instance_id出现三四次执行记录数据库最终状态是失败的但实际业务操作执行了多次。这个问题靠两层解决一是worker侧加了一个去重表记录已处理的instance_id重复消息直接丢弃二是重试扫描器只在instance表状态为“待执行”时才投递防止重试阶段和消费阶段互相打架。4.3 运维视角监控、告警与死信处理调度系统最怕的不是出故障而是出故障了没人知道。ax上线时我就把监控体系一起搭了核心指标就四个调度延迟任务触发时间与实际投递时间之差超过5秒就要关注队列积压Redis Stream xlen长度超过10000报警执行成功率最近一小时的执行成功百分比低于95%报警执行器心跳某个执行器离线超过15秒报警告警渠道直接接的飞书机器人告警级别分P1和P2P1是任务大面积失败或队列严重堆积需要立刻处理P2是单个任务连续失败在死信表里待人工确认的条数超过阈值。死信表的设计也很重要。ax_task_dead_letter表结构比实例表多一个solution字段人工介入时填写处理方案和结果。每周我会看一遍死信表从中能发现很多周期性问题的规律比如某个第三方接口每周三凌晨必挂后来才发现是那个服务商每周三发版。5. 常见问题与排查技巧实录5.1 常见问题速查表把这几年来群里被问得最多的问题整理一遍按表现症状分类排查方法直接可抄症状可能原因排查步骤解决方案任务到点不触发调度器进程挂了或扫表SQL异常看调度器日志、检查进程存活配置守护进程自动拉起异常要报警任务重复执行biz_id生成规则碰撞或pending重复消费查实例表是否有多个记录看日志中instance_id是否重复修复biz_id生成规则worker加去重缓存任务一直卡在RUNNING业务回调超时设置过长或回调地址不可达查看worker当前协程数、业务日志缩短超时时间到30秒确认回调地址worker重启后任务丢失消息在Stream里但worker没有xack查pending队列长度和具体消息增加启动时的pending消息扫描逻辑延迟任务执行不准时调度器扫描周期过长或投递后排队看调度延迟指标、队列积压长度调小SCAN_INTERVAL增加执行器节点Redis内存暴涨Stream消息积压且无人消费看xlen和内存监控确认消费者组是否活跃必要时清理积压5.2 一分钟定位任务问题的小技巧最后分享一个实操中屡试不爽的排查方法。当你看到某个任务异常不要先翻代码直接查这个任务的instance_id然后把任务实例表、Redis Stream里的消息、执行器日志、业务回调日志四条线的信息串起来按时间线对一遍基本能在一分钟内确定问题出在哪个环节。具体操作是用任务code查最近10条实例记录看状态变化的时间点用instance_id去Redis Stream的PEL里找消息是否存在再到执行器机器上grep这个instance_id的日志最后到业务服务里grep回调参数。这四步走完80%的问题都能定位到出错的层。做调度系统很残酷——它平时默默无闻一旦出问题就是全局性的所以排查套路一定要固化下来别临时抱佛脚。我自己做了这么多年最大的体会是调度系统不是一个“做完就完事”的功能模块它是一个需要持续观察和调整的基础设施。参数不是设一次就一劳永逸的任务量在涨、业务方在增加新需求、执行器节点在变调度策略就得跟着演进。如果你准备自研一套类似的东西先把任务模型定义清楚把调度和执行拆开把重试和幂等做扎实剩下的参数调优都可以在生产环境慢慢打磨。这套ax的经验希望能让你少踩几个我当年踩过的坑。
