分布式定时任务框架、负载均衡、OpenAPI异步调用这三个词叠在一起很多人第一反应就是“直接用 XXL-Job 不就行了”。但如果你遇到的是这样的场景外部业务方要通过开放接口动态提交定时任务平台负责把任务分发给不同的执行节点并且对外只能走异步语义不能长时间占用 HTTP 连接那你大概率会发现现成框架反而成了瓶颈。我就在这种背景下从零写了一套轻量的分布式定时任务框架里面包含调度中心、执行器、Redis 分布式锁、等开销负载均衡策略以及一整套 OpenAPI 异步创建任务和查询状态的实现。这篇文章把完整的思路、核心代码和踩坑记录整理出来适合已经用过 Spring Boot、Redis、MySQL想自己动手搞明白分布式调度底层逻辑的开发者。1. 为什么要自己写一个分布式定时任务框架1.1 业务里被单机调度“卡”住的场景公司的老系统里定时任务一直是写在一个单体应用里的。最初只有几个报表任务用 Spring 的Scheduled加一个线程池就能跑。但是随着业务方变多任务数量从十几个涨到几百个这时候单机调度的问题就非常明显了任务多了线程池排队一个任务执行时间过长会把后面的任务全部拖死服务重启或者宕机所有定时任务跟着一起没了线上部署多台实例时同一个Scheduled又会在每个实例上各执行一遍报表重复生成、短信重复发送。后来尝试过引入成熟框架但发现两个问题。第一这些框架本身是面向“运维配置任务”设计的界面和调度能力很强可我们的业务方希望直接通过 API 创建定时任务要求提交后立刻返回而任务到底执行成不成功由平台通过回调或者查询接口再告诉他们。这套“异步 OpenAPI”语义在现成框架里大部分是缺少的需要大量二次开发。第二我们并不需要极端的调度能力更核心的是要把任务均匀地分发到不同执行节点以及保证同一个任务在集群里不会同时跑多遍。与其被框架的约定绑住不如自己写一个贴合业务的小框架。1.2 自研框架的三大核心诉求动手之前我把需求收敛成三件事。第一任务调度必须是分布式的。多台调度实例同时在线但同一时刻一个任务只能被一个调度器领走。实现上不能靠“随机抢”要用分布式锁保证调度动作只有一个执行者。第二任务要能被负载均衡地分发到多个执行器。执行器节点注册上来之后调度器要根据每个节点当前的“开销”而不是简单轮询来选择目标。这里我实现了一个等开销负载均衡策略综合活跃任务数、历史响应耗时、CPU 空闲率来算权重让每个节点的实际负载尽量接近。第三外部调用必须异步化。OpenAPI 入口接收到创建任务请求后立刻落库并返回任务 ID真正执行放在调度链路里。调用方用任务 ID 查询状态或者等平台执行完成后回调通知。异步的好处很直接——调用方不用为一个可能执行几分钟的任务一直阻塞连接调度平台也不会因为大量慢任务被打垮。2. 整体架构和关键设计调度中心、执行器、分布式锁怎么配合2.1 模块划分和一次完整调度流程整个框架从物理上分成两类进程调度中心Dispatcher和执行器Worker。调度中心可以部署多台执行器也可以部署多台调度中心通过心跳判断执行器是否存活。存储层是 MySQL 加 RedisMySQL 持久化任务定义和每次执行日志Redis 用来做分布式锁、执行器心跳和临时状态缓存。一次完整流程是这样走的业务方调用 OpenAPI 创建任务接口写入job_info表返回jobId。调度中心每隔一秒去 MySQL 扫描next_trigger_time now且状态为等待中的任务。调度中心先抢 Redis 锁抢到锁的实例才有资格执行本次扫描和分发避免多实例重复调度。调度中心根据负载均衡策略从存活的 worker 列表里选一个节点。调度中心把任务标记为“已分发”并调用 worker 的/run接口。worker 接到任务后丢进本地线程池执行执行结束后把结果回报给调度中心写入job_log。调用方通过 OpenAPI 查询任务状态或者接收执行完成后的回调。这套设计里最关键的就是第 4 步和第 5 步中间的“状态流转”。如果不小心两个调度实例会同时发现同一个到期任务同时去调用 worker任务就重复执行了。所以我在分发之前加了两道保险Redis 分布式锁负责“调度动作互斥”数据库乐观锁负责“任务状态原子变更”。2.2 Redis 分布式锁调度防重和任务分发的基石分布式锁在分布式定时任务里扮演的角色先想清楚一个问题为什么不能只用数据库的唯一索引因为调度中心只是“发现任务”和“分发任务”任务真正的执行在 worker 侧数据库层面只能保证任务不被重复领取但不能保证两个调度中心不会同时把同一个任务发给两个不同的 worker。所以调度动作本身必须互斥。我的第一版实现很简单Scheduled(fixedDelay 1000) public void scheduleDispatch() { // 先尝试加锁如果加不上说明有别的调度实例在干活 Boolean locked redisTemplate.opsForValue() .setIfAbsent(lock:dispatch, localIp, Duration.ofSeconds(10)); if (Boolean.FALSE.equals(locked)) { return; } try { doDispatch(); } finally { redisTemplate.delete(lock:dispatch); } }setIfAbsent对应 Redis 的SET NX设置过期时间对应PX。这样即使某个调度实例在干活中途宕机锁也会在 10 秒后自动释放其他实例可以接管。但这样写有一个隐患如果doDispatch()执行时间超过 10 秒锁会提前过期另一个调度实例就会冲进来两个实例同时分发。我后来在锁里加了一个“续期线程”每 3 秒检查一次锁是否还在自己手里是的话就把过期时间重置为 10 秒。这个逻辑和 Redisson 的看门狗机制是一致的。生产上可以直接用 Redisson 的RLock但为了把原理讲透我建议自己实现一遍。2.3 负载均衡策略从轮询到等开销负载均衡执行器节点多了以后负载均不均匀直接决定任务能不能及时跑完。最朴素的轮询策略在任务耗时差不多的情况下是够用的可一旦出现某个节点正在跑一个跑半个小时的报表任务另一个节点全都空闲轮询还是会把新任务继续塞给繁忙节点导致整体吞吐直线下降。所以我实现了“等开销负载均衡”。这个“等开销”是我自己的叫法意思是经过负载均衡之后每个节点的综合开销水平应当尽量相等而不是简单地任务数量一样多。实现上给每个 worker 算一个综合负载值当前活跃任务数activeCount这个值从 worker 心跳里带上来的。最近 5 分钟任务的平均耗时avgCostTime。节点上报的 CPU 空闲率cpuIdlePercent。得到负载指数load activeCount * 10 avgCostTime / 1000 (100 - cpuIdlePercent) / 10。调度时选load最小的节点。这样做的好处是一个节点就算当前任务数少但如果它上报的 CPU 已经跑得很满也不会被优先选中。这里有个细节worker 上报指标是有延迟的调度中心拿到的可能是几秒前的状态。所以等开销负载均衡在任务并发不高的时候非常稳定但瞬时大量任务涌入时还是可能倾斜。为了弥补我额外加了并发控制worker 本地线程池的队列长度一旦超过阈值会主动拒绝任务并要求调度中心换一个节点。3. 从零到一实现调度中心、执行器、OpenAPI异步调用3.1 存储设计与落库 SQL一套分布式定时任务框架最需要持久化的是三块内容任务定义、执行日志、worker 节点信息。下面是核心表结构。CREATE TABLE job_info ( id bigint NOT NULL AUTO_INCREMENT, job_name varchar(128) NOT NULL, job_handler varchar(255) NOT NULL COMMENT 任务处理器的bean名称, params text COMMENT 业务参数, cron_expr varchar(64) DEFAULT NULL COMMENT cron表达式为空表示一次性任务, next_trigger_time bigint DEFAULT NULL COMMENT 下一次触发时间戳(ms), status tinyint NOT NULL DEFAULT 0 COMMENT 0等待中 1运行中 2成功 3失败 4已暂停, owner_worker_id varchar(64) DEFAULT NULL COMMENT 当前负责执行的worker, version int NOT NULL DEFAULT 0 COMMENT 乐观锁版本号, callback_url varchar(512) DEFAULT NULL COMMENT 执行完成后的回调地址, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_status_next_time (status, next_trigger_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE job_log ( id bigint NOT NULL AUTO_INCREMENT, job_id bigint NOT NULL, worker_id varchar(64) NOT NULL, status tinyint NOT NULL COMMENT 执行状态: 0执行中 1成功 2失败, message text, start_time bigint NOT NULL, end_time bigint DEFAULT NULL, PRIMARY KEY (id), KEY idx_job_id (job_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE worker_node ( id varchar(64) NOT NULL COMMENT worker唯一标识, host varchar(64) NOT NULL, port int NOT NULL, active_count int DEFAULT 0, cpu_idle_percent int DEFAULT 100, last_heartbeat_time bigint NOT NULL, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;job_info里的version字段很关键。调度中心分发任务时不是直接更新状态而是执行类似UPDATE job_info SET status1, versionversion1 WHERE id? AND status0 AND version?的语句。如果影响行数为 0说明任务已经被别的调度器领走了当前调度器只能放弃。这个乐观锁配合 Redis 锁是双层防护。3.2 调度主循环与分布式锁落地调度中心的主循环我用 Spring Boot 的Scheduled来触发每秒跑一次。每次只处理一批最早到期的任务避免一次扫描把数据库打满。核心流程public void doDispatch() { // 扫描未来10秒内需要执行的任务避免一条条绑时间点 long now System.currentTimeMillis(); ListJobInfo dueJobs jobInfoMapper.findJobsByStatusAndTime(0, now, now 10000); for (JobInfo job : dueJobs) { distributeJob(job); } // 把已经到期的cron任务计算下次触发时间 ListJobInfo cronJobs jobInfoMapper.findCronJobs(); for (JobInfo job : cronJobs) { if (job.getNextTriggerTime() now) { updateNextTriggerTimeAndDispatch(job); } } }distributeJob里要做的事情比较多。public boolean distributeJob(JobInfo job) { // 乐观锁占用任务状态 int rows jobInfoMapper.tryClaim(job.getId(), localIp, job.getVersion()); if (rows 0) { return false; } // 选择负载最低的 worker WorkerNode worker loadBalancer.choose(workerRegistry.getAliveWorkers()); if (worker null) { // 没有可用worker把任务重新置回等待中留待下次调度 jobInfoMapper.resetStatus(job.getId()); return false; } try { restTemplate.postForEntity(http:// worker.getHost() : worker.getPort() /run, new RunRequest(job.getId(), job.getJobHandler(), job.getParams()), String.class); return true; } catch (Exception e) { // 调用失败释放任务状态下一次调度会再试 jobInfoMapper.resetStatus(job.getId()); return false; } }tryClaim的 SQL 是这样的UPDATE job_info SET status 1, owner_worker_id #{workerId}, version version 1 WHERE id #{id} AND status 0 AND version #{version}这里有个细节我一开始忽略了调用 worker 的/run是一个网络请求请求发出去了不代表 worker 真的开始执行。假如 worker 收到了请求但还没来得及更新自己的状态然后宕机了调度中心这边可能认为分发失败并把任务重新置回等待中。这就有可能出现两次执行。要彻底解决这个问题还得引入任务执行状态确认机制worker 收到任务后先落一条“已接收”日志调度中心收到确认后不再处理。不过这会增加复杂度在简单场景里我选择了“最多超时重试一次”的策略配合 worker 端的 Redis 幂等标识来兜底后文会详细说明。3.3 执行器实现与心跳上报执行器本质上是一个独立的 Spring Boot 服务暴露两个 HTTP 接口/run接收任务/heartbeat上报状态。/run接口的核心逻辑PostMapping(/run) public ResponseEntityString run(RequestBody RunRequest request) { // 幂等校验同一jobId同一时间戳只允许执行一次 String idempotentKey request.getJobId() : request.getTriggerTimestamp(); Boolean isNew stringRedisTemplate.opsForValue() .setIfAbsent(idempotent: idempotentKey, 1, Duration.ofHours(1)); if (!Boolean.TRUE.equals(isNew)) { return ResponseEntity.ok(duplicated); } // 交给本地线程池异步执行 taskExecutor.execute(() - { JobLog log new JobLog(); log.setJobId(request.getJobId()); log.setWorkerId(workerId); log.setStatus(0); log.setStartTime(System.currentTimeMillis()); jobLogMapper.insert(log); try { // 通过Spring上下文获取对应的Handler JobHandler handler context.getBean(request.getJobHandler(), JobHandler.class); handler.handle(request.getParams()); jobLogMapper.updateStatus(log.getId(), 1); // 执行成功回调通知 notifyCallback(request, true, null); } catch (Exception e) { jobLogMapper.updateStatus(log.getId(), 2, e.getMessage()); notifyCallback(request, false, e.getMessage()); } }); return ResponseEntity.ok(accepted); }心跳上报是另一个定时任务每 5 秒一次。上报内容包含当前活跃任务数、CPU 空闲率、最近任务平均耗时。调度中心收到心跳后更新worker_node表同时判断如果超过 15 秒没收到心跳就把这个 worker 标记为离线调度任务时自动跳过。3.4 负载均衡算法代码实现等开销负载均衡的核心类我写成这样Component public class LoadBalancer { private final WorkerRegistry registry; public WorkerNode choose() { ListWorkerNode alive registry.getAliveWorkers(); if (alive.isEmpty()) { return null; } return alive.stream() .min(Comparator.comparingDouble(this::calcLoad)) .orElse(null); } private double calcLoad(WorkerNode worker) { // 活跃任务数权重最高 double activeScore worker.getActiveCount() * 10; // 平均耗时单位毫秒换算成加权分 double avgCostScore worker.getAvgCostTime() / 1000.0; // CPU空闲率越低得分越高 double cpuScore (100 - worker.getCpuIdlePercent()) / 10.0; return activeScore avgCostScore cpuScore; } }在实际测试里这个策略起效的关键在于 activeCount 的准确性。worker 执行任务前要把计数器加一任务结束后减一同时把本次耗时记录到滑动窗口。如果少维护了计数器负载均衡就退化成随机分发。除了等开销策略我还保留了轮询和随机策略。通过配置中心切换策略方便对比效果。如果你不想自己维护节点心跳也可以用 Redis 的ZSet按心跳时间戳排序来管理存活节点调度前过滤掉过期节点省一张数据库表代价是查询不够灵活。3.5 OpenAPI异步创建任务与状态查询OpenAPI 的设计核心是“异步”。调用方创建一个任务正常情况下不应该等待任务执行完成而是立刻得到一个任务 ID。然后调用方循环查询状态或者由平台在执行完成时回调调用方。我的 OpenAPI 接口长这样RestController RequestMapping(/openapi) public class OpenApiController { PostMapping(/jobs) public ResponseEntityCreateJobResponse createJob(RequestBody CreateJobRequest request) { // 幂等处理同一个clientRequestId不重复创建 String idempotentKey openapi:job: request.getClientRequestId(); Boolean isNew stringRedisTemplate.opsForValue() .setIfAbsent(idempotentKey, 1, Duration.ofDays(1)); if (!Boolean.TRUE.equals(isNew)) { JobInfo existing jobInfoMapper.findByClientRequestId(request.getClientRequestId()); return ResponseEntity.ok(new CreateJobResponse(existing.getId(), existing.getStatus())); } // 先落库再返回 JobInfo job new JobInfo(); job.setJobName(request.getJobName()); job.setJobHandler(request.getJobHandler()); job.setParams(request.getParams()); job.setCronExpr(request.getCronExpr()); job.setNextTriggerTime(calculateNextTriggerTime(request)); job.setCallbackUrl(request.getCallbackUrl()); job.setStatus(0); jobInfoMapper.insert(job); return ResponseEntity.ok(new CreateJobResponse(job.getId(), job.getStatus())); } GetMapping(/jobs/{id}) public ResponseEntityJobStatusResponse getJobStatus(PathVariable(id) Long id) { JobInfo job jobInfoMapper.findById(id); return ResponseEntity.ok(new JobStatusResponse(job.getId(), job.getStatus(), job.getParams())); } }创建接口的时序很重要必须先落库再返回一个“已接收”的响应。如果先返回调用方立刻查状态会查不到这条记录。这里可以做成事务保证插入和返回之间没有中间状态。有些业务方希望任务完成后平台主动通知他们这时候可以用callbackUrl。在 worker 执行完任务后由调度中心回调外部系统的接口回调体里带上任务 ID、执行状态、错误信息。要注意的是回调接口必须是幂等的重试时不能造成业务重复。我在回调请求里带了一个executionId每次执行生成一个 UUID回调接收方可以根据这个字段去重。3.6 异步任务如何做到可追踪很多人在自研时忽略一个点异步任务虽然返回快但调用方怎么知道任务到底跑没跑如果只提供一个“查状态”的接口调用方就得不停轮询体验很差。我采用的是“状态流转变更记录”的方式job_info表里的状态从 0 到 1 到 2 或 3每一步都在job_log里写一条记录。调用方查询时不仅可以拿到任务当前状态还能拿到每一步的执行时间和错误信息。对于长时间不回调的任务我在调度中心加了一个“超时扫描”。比如某个一次性任务分发出去后 10 分钟还没从status1变成 2 或 3就视为执行异常把状态重置为 0 并重新调度。这个机制既保护了外部调用方体验也避免任务因为 worker 异常卡死。4. 上线以来踩过的坑重复调度、锁超时、任务丢失4.1 重复执行问题排查与幂等设计第一版上线后最严重的问题就是任务偶尔重复执行。不是每一条都重复而是集中在任务执行时间比较长、调度中心刚好发生集群扩容的时候。排查了很久发现问题出在 Redis 锁上调度实例 A 抢到锁正在逐个分发任务某一次分发时调用 worker 超时A 在异常处理时把锁释放了这时候调度实例 B 抢到锁把 A 已经分发过的任务又分发了一遍。worker 端收到两个一模一样的执行请求。光靠 Redis 锁根本防不住这种“跨调度周期”的重复。所以我在 worker 端加了幂等对同一任务同一次触发只执行一次用jobId 触发时间戳作为 Redis 幂等键。但这里也有坑如果 worker 在执行完任务之后、更新状态之前重启了Redis 里的幂等键还在任务就不会被重新执行造成任务丢失。因此我又在调度中心加了一个超时重置逻辑超过一定时间没收到执行结果删除幂等键并重新分发。最终总结下来双重幂等是最稳的调度中心分发前用数据库乐观锁保证状态流转worker 执行前用 Redis 幂等键保证单次执行不被重复。两者缺一不可。4.2 worker 宕机后的任务拉起有一次我做了个破坏性测试正在执行一个耗时 5 分钟的任务直接 kill 掉 worker。任务卡在status1状态调度中心没有感知因为心跳数据是每 5 秒上报一次worker 没了之后调度中心要最多等 15 秒才能判断它离线。离线判断可以做但卡在运行中的任务怎么恢复我在调度中心加了一个“孤儿任务扫描”SELECT * FROM job_info WHERE status 1 AND owner_worker_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM worker_node WHERE worker_node.id job_info.owner_worker_id AND worker_node.last_heartbeat_time #{now - 15000} )把这些任务找出来后把状态重置为 0并清掉 worker 端的幂等键。这里有一个非常容易被忽略的细节重置任务前要确认旧 worker 确实已经死透了。如果旧 worker 只是网络抖动任务还在它上面执行你把状态重置了另一个 worker 也会执行同一个任务两边同时操作业务数据后果不堪设想。因此我坚持用“心跳超时 TCP 探测”双重判断只有 TCP 连接也连不上才认为 worker 已经离线。4.3 Redis 锁超时和续约问题前面提到过分布式锁如果过期时间设短了任务还没分完锁就自动释放会造成并发调度如果设长了调度实例宕机后其他实例要等很久才能接管。我最初设了 10 秒但是想象一下这个场景数据库里有几千个到期任务调度中心在逐条分发每一条都要走一次网络请求10 秒根本不够。于是我用“续期线程”定时给 Redis 锁续期相当于实现了一个简易看门狗。续期线程实现ScheduledExecutorService renewThread Executors.newSingleThreadScheduledExecutor(r - { Thread t new Thread(r, lock-renew); t.setDaemon(true); return t; }); renewThread.scheduleAtFixedRate(() - { String currentValue redisTemplate.opsForValue().get(lock:dispatch); if (localIp.equals(currentValue)) { redisTemplate.expire(lock:dispatch, Duration.ofSeconds(10)); } }, 0, 3, TimeUnit.SECONDS);但注意如果调度实例长时间持有锁不释放其他实例会一直空转。所以我在finally块里释放锁之前还会判断锁的 value 是否还是自己的 IP防止把别人的锁误删。这个细节建议直接参照 Redisson 的源码实现自己造的轮子稳定性确实不如成熟库。4.4 OpenAPI 异步接口的事务与幂等坑OpenAPI 异步接口最容易出的问题有两个。一个是事务问题。我一开始创建任务和计算下次触发时间是在同一个事务里但因为业务方通过 OpenAPI 创建任务的频率很高数据库连接经常不够用偶尔会出现任务插入后事务没提交线程池已经调度到了这条数据的情况。这会导致调度器拿到的是“半条”数据比如jobHandler字段还是 null。解决方法是把“创建任务”和“任务调度”彻底解耦插入完成后直接提交事务调度线程下一次扫描时自然能看到完整数据。另一个是重复创建。调用方可能因为网络重试把同一个创建请求发两次。我在接口入口用clientRequestId做幂等第一次请求把clientRequestId存入 Redis后续相同请求直接返回已有任务。这里要注意 Redis 的setIfAbsent和业务落库必须保证最终数据一致。如果客户端发的是同一个请求但参数不同幂等键只能拦截完全相同的情况参数不同会直接报错。5. 压测评估和后续扩展方向5.1 小规模压测实测结果整个框架在一个三节点的测试环境里跑了一轮一个调度中心两个 worker配置都是 4C8G。准备 1000 个一次性任务全部通过 OpenAPI 批量创建然后观察调度链路。实测结果是完整的任务吞吐在每分钟 2500 个左右超过这个量之后 worker 线程池开始排队接口响应变慢但不会丢任务。单任务从创建到执行的延迟基本在 1 秒到 2 秒之间因为调度主循环是每秒扫一次所以这个延迟是可预期的。压测中我发现job_info表的状态索引特别重要。如果没有idx_status_next_time扫描到期任务会导致全表扫描任务量超过万级以后调度延迟会直线上升。加了这个组合索引后扫描基本稳定在 10 毫秒内。5.2 后续可以怎么扩展这套框架目前最大的短板是缺少停机维护时的优雅处理。worker 在接受新任务前可以先把自己的状态置为“维护中”让调度中心不再分配新任务等本地线程池跑完再真正停机。另外失败重试目前只做了简单的“重置状态重新调度”对于重试次数和重试间隔还没做配置化。后续计划引入几种策略增强机制一是适配不同优先级任务让高优任务可以抢占低优任务二是接入 Prometheus 指标把调度延迟、任务成功率、worker 负载情况可视化三是把 OpenAPI 的异步回调做成可重试的消息队列避免回调 HTTP 接口临时抖动导致通知丢失。另一个我强烈建议的扩展方向是执行器端的优雅停机。demo 版本里我把任务丢进线程池后进程如果杀掉了执行中的任务直接丢失。可以在项目里维护一个“正在执行任务集合”在收到停机信号时拒绝新任务并等待已有任务超时或者上报中断状态。最后分享一个实际运维中的小技巧worker 和调度中心的时间一定要做 NTP 同步。调度算法里大量时间戳用的是毫秒如果机器之间时钟偏差太大就可能出现任务刚创建出来就立刻触发或者调度中心判断 worker 心跳超时的误报。别小看这个细节我第一版上线时唯一一次线上事故就是时钟偏差导致的。分布式定时任务框架自己完整实现一遍最大的收获不是代码能跑而是彻底理解了调度、锁、负载均衡这三者在分布式环境下的协作方式。接下来再碰到任何调度类的业务需求第一反应就不再是盲目堆中间件而是能根据业务流量和可靠性要求随时画出一套适合自己的方案。
分布式定时任务框架、负载均衡、OpenAPI异步调用这三个词叠在一起很多人第一反应就是“直接用 XXL-Job 不就行了”。但如果你遇到的是这样的场景外部业务方要通过开放接口动态提交定时任务平台负责把任务分发给不同的执行节点并且对外只能走异步语义不能长时间占用 HTTP 连接那你大概率会发现现成框架反而成了瓶颈。我就在这种背景下从零写了一套轻量的分布式定时任务框架里面包含调度中心、执行器、Redis 分布式锁、等开销负载均衡策略以及一整套 OpenAPI 异步创建任务和查询状态的实现。这篇文章把完整的思路、核心代码和踩坑记录整理出来适合已经用过 Spring Boot、Redis、MySQL想自己动手搞明白分布式调度底层逻辑的开发者。1. 为什么要自己写一个分布式定时任务框架1.1 业务里被单机调度“卡”住的场景公司的老系统里定时任务一直是写在一个单体应用里的。最初只有几个报表任务用 Spring 的Scheduled加一个线程池就能跑。但是随着业务方变多任务数量从十几个涨到几百个这时候单机调度的问题就非常明显了任务多了线程池排队一个任务执行时间过长会把后面的任务全部拖死服务重启或者宕机所有定时任务跟着一起没了线上部署多台实例时同一个Scheduled又会在每个实例上各执行一遍报表重复生成、短信重复发送。后来尝试过引入成熟框架但发现两个问题。第一这些框架本身是面向“运维配置任务”设计的界面和调度能力很强可我们的业务方希望直接通过 API 创建定时任务要求提交后立刻返回而任务到底执行成不成功由平台通过回调或者查询接口再告诉他们。这套“异步 OpenAPI”语义在现成框架里大部分是缺少的需要大量二次开发。第二我们并不需要极端的调度能力更核心的是要把任务均匀地分发到不同执行节点以及保证同一个任务在集群里不会同时跑多遍。与其被框架的约定绑住不如自己写一个贴合业务的小框架。1.2 自研框架的三大核心诉求动手之前我把需求收敛成三件事。第一任务调度必须是分布式的。多台调度实例同时在线但同一时刻一个任务只能被一个调度器领走。实现上不能靠“随机抢”要用分布式锁保证调度动作只有一个执行者。第二任务要能被负载均衡地分发到多个执行器。执行器节点注册上来之后调度器要根据每个节点当前的“开销”而不是简单轮询来选择目标。这里我实现了一个等开销负载均衡策略综合活跃任务数、历史响应耗时、CPU 空闲率来算权重让每个节点的实际负载尽量接近。第三外部调用必须异步化。OpenAPI 入口接收到创建任务请求后立刻落库并返回任务 ID真正执行放在调度链路里。调用方用任务 ID 查询状态或者等平台执行完成后回调通知。异步的好处很直接——调用方不用为一个可能执行几分钟的任务一直阻塞连接调度平台也不会因为大量慢任务被打垮。2. 整体架构和关键设计调度中心、执行器、分布式锁怎么配合2.1 模块划分和一次完整调度流程整个框架从物理上分成两类进程调度中心Dispatcher和执行器Worker。调度中心可以部署多台执行器也可以部署多台调度中心通过心跳判断执行器是否存活。存储层是 MySQL 加 RedisMySQL 持久化任务定义和每次执行日志Redis 用来做分布式锁、执行器心跳和临时状态缓存。一次完整流程是这样走的业务方调用 OpenAPI 创建任务接口写入job_info表返回jobId。调度中心每隔一秒去 MySQL 扫描next_trigger_time now且状态为等待中的任务。调度中心先抢 Redis 锁抢到锁的实例才有资格执行本次扫描和分发避免多实例重复调度。调度中心根据负载均衡策略从存活的 worker 列表里选一个节点。调度中心把任务标记为“已分发”并调用 worker 的/run接口。worker 接到任务后丢进本地线程池执行执行结束后把结果回报给调度中心写入job_log。调用方通过 OpenAPI 查询任务状态或者接收执行完成后的回调。这套设计里最关键的就是第 4 步和第 5 步中间的“状态流转”。如果不小心两个调度实例会同时发现同一个到期任务同时去调用 worker任务就重复执行了。所以我在分发之前加了两道保险Redis 分布式锁负责“调度动作互斥”数据库乐观锁负责“任务状态原子变更”。2.2 Redis 分布式锁调度防重和任务分发的基石分布式锁在分布式定时任务里扮演的角色先想清楚一个问题为什么不能只用数据库的唯一索引因为调度中心只是“发现任务”和“分发任务”任务真正的执行在 worker 侧数据库层面只能保证任务不被重复领取但不能保证两个调度中心不会同时把同一个任务发给两个不同的 worker。所以调度动作本身必须互斥。我的第一版实现很简单Scheduled(fixedDelay 1000) public void scheduleDispatch() { // 先尝试加锁如果加不上说明有别的调度实例在干活 Boolean locked redisTemplate.opsForValue() .setIfAbsent(lock:dispatch, localIp, Duration.ofSeconds(10)); if (Boolean.FALSE.equals(locked)) { return; } try { doDispatch(); } finally { redisTemplate.delete(lock:dispatch); } }setIfAbsent对应 Redis 的SET NX设置过期时间对应PX。这样即使某个调度实例在干活中途宕机锁也会在 10 秒后自动释放其他实例可以接管。但这样写有一个隐患如果doDispatch()执行时间超过 10 秒锁会提前过期另一个调度实例就会冲进来两个实例同时分发。我后来在锁里加了一个“续期线程”每 3 秒检查一次锁是否还在自己手里是的话就把过期时间重置为 10 秒。这个逻辑和 Redisson 的看门狗机制是一致的。生产上可以直接用 Redisson 的RLock但为了把原理讲透我建议自己实现一遍。2.3 负载均衡策略从轮询到等开销负载均衡执行器节点多了以后负载均不均匀直接决定任务能不能及时跑完。最朴素的轮询策略在任务耗时差不多的情况下是够用的可一旦出现某个节点正在跑一个跑半个小时的报表任务另一个节点全都空闲轮询还是会把新任务继续塞给繁忙节点导致整体吞吐直线下降。所以我实现了“等开销负载均衡”。这个“等开销”是我自己的叫法意思是经过负载均衡之后每个节点的综合开销水平应当尽量相等而不是简单地任务数量一样多。实现上给每个 worker 算一个综合负载值当前活跃任务数activeCount这个值从 worker 心跳里带上来的。最近 5 分钟任务的平均耗时avgCostTime。节点上报的 CPU 空闲率cpuIdlePercent。得到负载指数load activeCount * 10 avgCostTime / 1000 (100 - cpuIdlePercent) / 10。调度时选load最小的节点。这样做的好处是一个节点就算当前任务数少但如果它上报的 CPU 已经跑得很满也不会被优先选中。这里有个细节worker 上报指标是有延迟的调度中心拿到的可能是几秒前的状态。所以等开销负载均衡在任务并发不高的时候非常稳定但瞬时大量任务涌入时还是可能倾斜。为了弥补我额外加了并发控制worker 本地线程池的队列长度一旦超过阈值会主动拒绝任务并要求调度中心换一个节点。3. 从零到一实现调度中心、执行器、OpenAPI异步调用3.1 存储设计与落库 SQL一套分布式定时任务框架最需要持久化的是三块内容任务定义、执行日志、worker 节点信息。下面是核心表结构。CREATE TABLE job_info ( id bigint NOT NULL AUTO_INCREMENT, job_name varchar(128) NOT NULL, job_handler varchar(255) NOT NULL COMMENT 任务处理器的bean名称, params text COMMENT 业务参数, cron_expr varchar(64) DEFAULT NULL COMMENT cron表达式为空表示一次性任务, next_trigger_time bigint DEFAULT NULL COMMENT 下一次触发时间戳(ms), status tinyint NOT NULL DEFAULT 0 COMMENT 0等待中 1运行中 2成功 3失败 4已暂停, owner_worker_id varchar(64) DEFAULT NULL COMMENT 当前负责执行的worker, version int NOT NULL DEFAULT 0 COMMENT 乐观锁版本号, callback_url varchar(512) DEFAULT NULL COMMENT 执行完成后的回调地址, create_time datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (id), KEY idx_status_next_time (status, next_trigger_time) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE job_log ( id bigint NOT NULL AUTO_INCREMENT, job_id bigint NOT NULL, worker_id varchar(64) NOT NULL, status tinyint NOT NULL COMMENT 执行状态: 0执行中 1成功 2失败, message text, start_time bigint NOT NULL, end_time bigint DEFAULT NULL, PRIMARY KEY (id), KEY idx_job_id (job_id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4; CREATE TABLE worker_node ( id varchar(64) NOT NULL COMMENT worker唯一标识, host varchar(64) NOT NULL, port int NOT NULL, active_count int DEFAULT 0, cpu_idle_percent int DEFAULT 100, last_heartbeat_time bigint NOT NULL, PRIMARY KEY (id) ) ENGINEInnoDB DEFAULT CHARSETutf8mb4;job_info里的version字段很关键。调度中心分发任务时不是直接更新状态而是执行类似UPDATE job_info SET status1, versionversion1 WHERE id? AND status0 AND version?的语句。如果影响行数为 0说明任务已经被别的调度器领走了当前调度器只能放弃。这个乐观锁配合 Redis 锁是双层防护。3.2 调度主循环与分布式锁落地调度中心的主循环我用 Spring Boot 的Scheduled来触发每秒跑一次。每次只处理一批最早到期的任务避免一次扫描把数据库打满。核心流程public void doDispatch() { // 扫描未来10秒内需要执行的任务避免一条条绑时间点 long now System.currentTimeMillis(); ListJobInfo dueJobs jobInfoMapper.findJobsByStatusAndTime(0, now, now 10000); for (JobInfo job : dueJobs) { distributeJob(job); } // 把已经到期的cron任务计算下次触发时间 ListJobInfo cronJobs jobInfoMapper.findCronJobs(); for (JobInfo job : cronJobs) { if (job.getNextTriggerTime() now) { updateNextTriggerTimeAndDispatch(job); } } }distributeJob里要做的事情比较多。public boolean distributeJob(JobInfo job) { // 乐观锁占用任务状态 int rows jobInfoMapper.tryClaim(job.getId(), localIp, job.getVersion()); if (rows 0) { return false; } // 选择负载最低的 worker WorkerNode worker loadBalancer.choose(workerRegistry.getAliveWorkers()); if (worker null) { // 没有可用worker把任务重新置回等待中留待下次调度 jobInfoMapper.resetStatus(job.getId()); return false; } try { restTemplate.postForEntity(http:// worker.getHost() : worker.getPort() /run, new RunRequest(job.getId(), job.getJobHandler(), job.getParams()), String.class); return true; } catch (Exception e) { // 调用失败释放任务状态下一次调度会再试 jobInfoMapper.resetStatus(job.getId()); return false; } }tryClaim的 SQL 是这样的UPDATE job_info SET status 1, owner_worker_id #{workerId}, version version 1 WHERE id #{id} AND status 0 AND version #{version}这里有个细节我一开始忽略了调用 worker 的/run是一个网络请求请求发出去了不代表 worker 真的开始执行。假如 worker 收到了请求但还没来得及更新自己的状态然后宕机了调度中心这边可能认为分发失败并把任务重新置回等待中。这就有可能出现两次执行。要彻底解决这个问题还得引入任务执行状态确认机制worker 收到任务后先落一条“已接收”日志调度中心收到确认后不再处理。不过这会增加复杂度在简单场景里我选择了“最多超时重试一次”的策略配合 worker 端的 Redis 幂等标识来兜底后文会详细说明。3.3 执行器实现与心跳上报执行器本质上是一个独立的 Spring Boot 服务暴露两个 HTTP 接口/run接收任务/heartbeat上报状态。/run接口的核心逻辑PostMapping(/run) public ResponseEntityString run(RequestBody RunRequest request) { // 幂等校验同一jobId同一时间戳只允许执行一次 String idempotentKey request.getJobId() : request.getTriggerTimestamp(); Boolean isNew stringRedisTemplate.opsForValue() .setIfAbsent(idempotent: idempotentKey, 1, Duration.ofHours(1)); if (!Boolean.TRUE.equals(isNew)) { return ResponseEntity.ok(duplicated); } // 交给本地线程池异步执行 taskExecutor.execute(() - { JobLog log new JobLog(); log.setJobId(request.getJobId()); log.setWorkerId(workerId); log.setStatus(0); log.setStartTime(System.currentTimeMillis()); jobLogMapper.insert(log); try { // 通过Spring上下文获取对应的Handler JobHandler handler context.getBean(request.getJobHandler(), JobHandler.class); handler.handle(request.getParams()); jobLogMapper.updateStatus(log.getId(), 1); // 执行成功回调通知 notifyCallback(request, true, null); } catch (Exception e) { jobLogMapper.updateStatus(log.getId(), 2, e.getMessage()); notifyCallback(request, false, e.getMessage()); } }); return ResponseEntity.ok(accepted); }心跳上报是另一个定时任务每 5 秒一次。上报内容包含当前活跃任务数、CPU 空闲率、最近任务平均耗时。调度中心收到心跳后更新worker_node表同时判断如果超过 15 秒没收到心跳就把这个 worker 标记为离线调度任务时自动跳过。3.4 负载均衡算法代码实现等开销负载均衡的核心类我写成这样Component public class LoadBalancer { private final WorkerRegistry registry; public WorkerNode choose() { ListWorkerNode alive registry.getAliveWorkers(); if (alive.isEmpty()) { return null; } return alive.stream() .min(Comparator.comparingDouble(this::calcLoad)) .orElse(null); } private double calcLoad(WorkerNode worker) { // 活跃任务数权重最高 double activeScore worker.getActiveCount() * 10; // 平均耗时单位毫秒换算成加权分 double avgCostScore worker.getAvgCostTime() / 1000.0; // CPU空闲率越低得分越高 double cpuScore (100 - worker.getCpuIdlePercent()) / 10.0; return activeScore avgCostScore cpuScore; } }在实际测试里这个策略起效的关键在于 activeCount 的准确性。worker 执行任务前要把计数器加一任务结束后减一同时把本次耗时记录到滑动窗口。如果少维护了计数器负载均衡就退化成随机分发。除了等开销策略我还保留了轮询和随机策略。通过配置中心切换策略方便对比效果。如果你不想自己维护节点心跳也可以用 Redis 的ZSet按心跳时间戳排序来管理存活节点调度前过滤掉过期节点省一张数据库表代价是查询不够灵活。3.5 OpenAPI异步创建任务与状态查询OpenAPI 的设计核心是“异步”。调用方创建一个任务正常情况下不应该等待任务执行完成而是立刻得到一个任务 ID。然后调用方循环查询状态或者由平台在执行完成时回调调用方。我的 OpenAPI 接口长这样RestController RequestMapping(/openapi) public class OpenApiController { PostMapping(/jobs) public ResponseEntityCreateJobResponse createJob(RequestBody CreateJobRequest request) { // 幂等处理同一个clientRequestId不重复创建 String idempotentKey openapi:job: request.getClientRequestId(); Boolean isNew stringRedisTemplate.opsForValue() .setIfAbsent(idempotentKey, 1, Duration.ofDays(1)); if (!Boolean.TRUE.equals(isNew)) { JobInfo existing jobInfoMapper.findByClientRequestId(request.getClientRequestId()); return ResponseEntity.ok(new CreateJobResponse(existing.getId(), existing.getStatus())); } // 先落库再返回 JobInfo job new JobInfo(); job.setJobName(request.getJobName()); job.setJobHandler(request.getJobHandler()); job.setParams(request.getParams()); job.setCronExpr(request.getCronExpr()); job.setNextTriggerTime(calculateNextTriggerTime(request)); job.setCallbackUrl(request.getCallbackUrl()); job.setStatus(0); jobInfoMapper.insert(job); return ResponseEntity.ok(new CreateJobResponse(job.getId(), job.getStatus())); } GetMapping(/jobs/{id}) public ResponseEntityJobStatusResponse getJobStatus(PathVariable(id) Long id) { JobInfo job jobInfoMapper.findById(id); return ResponseEntity.ok(new JobStatusResponse(job.getId(), job.getStatus(), job.getParams())); } }创建接口的时序很重要必须先落库再返回一个“已接收”的响应。如果先返回调用方立刻查状态会查不到这条记录。这里可以做成事务保证插入和返回之间没有中间状态。有些业务方希望任务完成后平台主动通知他们这时候可以用callbackUrl。在 worker 执行完任务后由调度中心回调外部系统的接口回调体里带上任务 ID、执行状态、错误信息。要注意的是回调接口必须是幂等的重试时不能造成业务重复。我在回调请求里带了一个executionId每次执行生成一个 UUID回调接收方可以根据这个字段去重。3.6 异步任务如何做到可追踪很多人在自研时忽略一个点异步任务虽然返回快但调用方怎么知道任务到底跑没跑如果只提供一个“查状态”的接口调用方就得不停轮询体验很差。我采用的是“状态流转变更记录”的方式job_info表里的状态从 0 到 1 到 2 或 3每一步都在job_log里写一条记录。调用方查询时不仅可以拿到任务当前状态还能拿到每一步的执行时间和错误信息。对于长时间不回调的任务我在调度中心加了一个“超时扫描”。比如某个一次性任务分发出去后 10 分钟还没从status1变成 2 或 3就视为执行异常把状态重置为 0 并重新调度。这个机制既保护了外部调用方体验也避免任务因为 worker 异常卡死。4. 上线以来踩过的坑重复调度、锁超时、任务丢失4.1 重复执行问题排查与幂等设计第一版上线后最严重的问题就是任务偶尔重复执行。不是每一条都重复而是集中在任务执行时间比较长、调度中心刚好发生集群扩容的时候。排查了很久发现问题出在 Redis 锁上调度实例 A 抢到锁正在逐个分发任务某一次分发时调用 worker 超时A 在异常处理时把锁释放了这时候调度实例 B 抢到锁把 A 已经分发过的任务又分发了一遍。worker 端收到两个一模一样的执行请求。光靠 Redis 锁根本防不住这种“跨调度周期”的重复。所以我在 worker 端加了幂等对同一任务同一次触发只执行一次用jobId 触发时间戳作为 Redis 幂等键。但这里也有坑如果 worker 在执行完任务之后、更新状态之前重启了Redis 里的幂等键还在任务就不会被重新执行造成任务丢失。因此我又在调度中心加了一个超时重置逻辑超过一定时间没收到执行结果删除幂等键并重新分发。最终总结下来双重幂等是最稳的调度中心分发前用数据库乐观锁保证状态流转worker 执行前用 Redis 幂等键保证单次执行不被重复。两者缺一不可。4.2 worker 宕机后的任务拉起有一次我做了个破坏性测试正在执行一个耗时 5 分钟的任务直接 kill 掉 worker。任务卡在status1状态调度中心没有感知因为心跳数据是每 5 秒上报一次worker 没了之后调度中心要最多等 15 秒才能判断它离线。离线判断可以做但卡在运行中的任务怎么恢复我在调度中心加了一个“孤儿任务扫描”SELECT * FROM job_info WHERE status 1 AND owner_worker_id IS NOT NULL AND NOT EXISTS ( SELECT 1 FROM worker_node WHERE worker_node.id job_info.owner_worker_id AND worker_node.last_heartbeat_time #{now - 15000} )把这些任务找出来后把状态重置为 0并清掉 worker 端的幂等键。这里有一个非常容易被忽略的细节重置任务前要确认旧 worker 确实已经死透了。如果旧 worker 只是网络抖动任务还在它上面执行你把状态重置了另一个 worker 也会执行同一个任务两边同时操作业务数据后果不堪设想。因此我坚持用“心跳超时 TCP 探测”双重判断只有 TCP 连接也连不上才认为 worker 已经离线。4.3 Redis 锁超时和续约问题前面提到过分布式锁如果过期时间设短了任务还没分完锁就自动释放会造成并发调度如果设长了调度实例宕机后其他实例要等很久才能接管。我最初设了 10 秒但是想象一下这个场景数据库里有几千个到期任务调度中心在逐条分发每一条都要走一次网络请求10 秒根本不够。于是我用“续期线程”定时给 Redis 锁续期相当于实现了一个简易看门狗。续期线程实现ScheduledExecutorService renewThread Executors.newSingleThreadScheduledExecutor(r - { Thread t new Thread(r, lock-renew); t.setDaemon(true); return t; }); renewThread.scheduleAtFixedRate(() - { String currentValue redisTemplate.opsForValue().get(lock:dispatch); if (localIp.equals(currentValue)) { redisTemplate.expire(lock:dispatch, Duration.ofSeconds(10)); } }, 0, 3, TimeUnit.SECONDS);但注意如果调度实例长时间持有锁不释放其他实例会一直空转。所以我在finally块里释放锁之前还会判断锁的 value 是否还是自己的 IP防止把别人的锁误删。这个细节建议直接参照 Redisson 的源码实现自己造的轮子稳定性确实不如成熟库。4.4 OpenAPI 异步接口的事务与幂等坑OpenAPI 异步接口最容易出的问题有两个。一个是事务问题。我一开始创建任务和计算下次触发时间是在同一个事务里但因为业务方通过 OpenAPI 创建任务的频率很高数据库连接经常不够用偶尔会出现任务插入后事务没提交线程池已经调度到了这条数据的情况。这会导致调度器拿到的是“半条”数据比如jobHandler字段还是 null。解决方法是把“创建任务”和“任务调度”彻底解耦插入完成后直接提交事务调度线程下一次扫描时自然能看到完整数据。另一个是重复创建。调用方可能因为网络重试把同一个创建请求发两次。我在接口入口用clientRequestId做幂等第一次请求把clientRequestId存入 Redis后续相同请求直接返回已有任务。这里要注意 Redis 的setIfAbsent和业务落库必须保证最终数据一致。如果客户端发的是同一个请求但参数不同幂等键只能拦截完全相同的情况参数不同会直接报错。5. 压测评估和后续扩展方向5.1 小规模压测实测结果整个框架在一个三节点的测试环境里跑了一轮一个调度中心两个 worker配置都是 4C8G。准备 1000 个一次性任务全部通过 OpenAPI 批量创建然后观察调度链路。实测结果是完整的任务吞吐在每分钟 2500 个左右超过这个量之后 worker 线程池开始排队接口响应变慢但不会丢任务。单任务从创建到执行的延迟基本在 1 秒到 2 秒之间因为调度主循环是每秒扫一次所以这个延迟是可预期的。压测中我发现job_info表的状态索引特别重要。如果没有idx_status_next_time扫描到期任务会导致全表扫描任务量超过万级以后调度延迟会直线上升。加了这个组合索引后扫描基本稳定在 10 毫秒内。5.2 后续可以怎么扩展这套框架目前最大的短板是缺少停机维护时的优雅处理。worker 在接受新任务前可以先把自己的状态置为“维护中”让调度中心不再分配新任务等本地线程池跑完再真正停机。另外失败重试目前只做了简单的“重置状态重新调度”对于重试次数和重试间隔还没做配置化。后续计划引入几种策略增强机制一是适配不同优先级任务让高优任务可以抢占低优任务二是接入 Prometheus 指标把调度延迟、任务成功率、worker 负载情况可视化三是把 OpenAPI 的异步回调做成可重试的消息队列避免回调 HTTP 接口临时抖动导致通知丢失。另一个我强烈建议的扩展方向是执行器端的优雅停机。demo 版本里我把任务丢进线程池后进程如果杀掉了执行中的任务直接丢失。可以在项目里维护一个“正在执行任务集合”在收到停机信号时拒绝新任务并等待已有任务超时或者上报中断状态。最后分享一个实际运维中的小技巧worker 和调度中心的时间一定要做 NTP 同步。调度算法里大量时间戳用的是毫秒如果机器之间时钟偏差太大就可能出现任务刚创建出来就立刻触发或者调度中心判断 worker 心跳超时的误报。别小看这个细节我第一版上线时唯一一次线上事故就是时钟偏差导致的。分布式定时任务框架自己完整实现一遍最大的收获不是代码能跑而是彻底理解了调度、锁、负载均衡这三者在分布式环境下的协作方式。接下来再碰到任何调度类的业务需求第一反应就不再是盲目堆中间件而是能根据业务流量和可靠性要求随时画出一套适合自己的方案。
