ElasticJob 的分片调度

一句话：ElasticJob 只负责把一次作业的 N 个分片项分配给当前在线的实例；每个实例拿到分片号后，自己决定处理哪些数据。它不替你路由数据，也不替你记录处理进度。

适用版本：ElasticJob 3.0.5（Apache ShardingSphere）、ZooKeeper 3.5+。3.x 的包名、接口签名和 Spring Boot 配置键与 2.x 不同，下文以 3.x 为准。

为什么需要

单机定时任务有三个硬伤：机器挂了任务就停；数据量涨了只能加线程，加不了机器；多实例部署时每个实例都会触发一次，同一批数据被重复处理。ElasticJob 把「一次执行」拆成 N 个分片项（sharding item），再把分片项分给在线实例。实例数可以小于、等于或大于分片数：小于时一个实例拿多项，大于时多出的实例空闲。扩缩容只改变「谁拿哪些项」，不改变「一共有多少项」。

核心机制

没有调度中心，协调靠 ZooKeeper

ElasticJob-Lite 是嵌入式框架，每个业务进程既是执行器也是调度器。实例启动后向 ZK 注册临时节点并参与作业级选主；Leader 在实例上下线时重新计算分片，把「分片项 → 实例」写进 ZK；所有实例监听映射，只跑分给自己的项。触发由各实例本地 cron 完成，没有中心节点下发。

所以 ElasticJob 的可用性直接依赖 ZK。连不上 ZK 的实例会停止执行作业，避免「Leader 已经重新分片、失联实例还在跑旧分片」的双写；连接恢复后作业自动继续，不需要重启应用。

框架分配的是分片号，不是数据

框架只知道「0 号分片归实例 A」，不知道订单 12345 落在哪个分片。取数规则必须由业务实现：

SELECT * FROM t_order
WHERE MOD(order_id, #{shardingTotalCount}) = #{shardingItem}
  AND status = 'PENDING';

hash(order_id) % n 中的 n 是作业配置的 shardingTotalCount，不是在线实例数。实例数变化只让分片项换宿主，不改变取数结果；分片总数变化才会改变路由，因此变更分片数必须配合数据迁移或全量重跑。

内置分片策略只有三种

策略	3.x 实现类	行为
平均	AverageAllocationJobShardingStrategy（默认）	尽量均分，余数从低序号实例开始加
奇偶	OdevitySortByNameJobShardingStrategy	按作业名哈希决定实例正序或倒序，再平均分配
轮询	RotateServerByNameJobShardingStrategy	按作业名哈希旋转实例列表，再平均分配

「按标签分配」不是内置策略。shardingItemParameters("0=shanghai,1=beijing") 只是给分片项附一个业务字符串，供 ShardingContext#getShardingParameter() 读取，不改变分配结果；地域亲和要自己实现策略。3.x 的扩展接口是 JobShardingStrategy：

public interface JobShardingStrategy extends TypedSPI {
    Map<JobInstance, List<Integer>> sharding(
            List<JobInstance> jobInstances, String jobName, int shardingTotalCount);
    String getType(); // 通过 jobShardingStrategyType("MY_STRATEGY") 选择
}

JobInstance 是实例对象，不是实例 ID 字符串。策略只在 Leader 上执行，结果由 Leader 写入 ZK。

故障接管：换宿主，不续进度

实例宕机后，ZK 临时节点在会话过期时删除，Leader 重新分片，把空缺项分给存活实例。ElasticJob 客户端默认请求 60s 会话超时；服务端按 2×tickTime ~ 20×tickTime 协商，默认 tickTime=2000ms 时上限为 40s。因此接管通常是几十秒级，不是 1–3 秒。

关键限制：ElasticJob 没有自动 checkpoint。DataflowJob 的 fetchData / processData 也不会把 offset 存到 ZK 或数据库。要从上次位置继续，必须由业务实现：处理完一批就更新业务表状态或 offset 表，下次从业务记录的位置继续。否则故障接管后，新实例会从分片起点重新扫，只能靠幂等兜底。流式模式的开关是 props.streaming.process=true（默认 false），fetchData 返回 null 或空集合才停止；处理完不更新状态，下一轮还会拉到同一批数据。

和 XXL-JOB 的关键差异

维度	ElasticJob 3.x	XXL-JOB 2.4.x
调度方式	各实例本地 cron，Leader 分片	Admin 统一触发，HTTP 调执行器
协调存储	ZooKeeper	MySQL + Admin
分片方式	固定 shardingTotalCount，Leader 分给实例	分片广播发给所有在线执行器，shardTotal 等于在线实例数
故障接管	会话过期后重新分片	心跳 30s，90s（30s×3）未更新才剔除；FAILOVER 在路由阶段探测存活实例
取数与进度	业务按 item/total 取数；无 checkpoint	业务按 index/total 取数；无 checkpoint

XXL-JOB 的分片广播不需要手工调整分片数：在线实例数变化后，下一次触发时 shardTotal 自然变化。两边都把「取哪些数据」留给业务代码。

实战场景

一、把分片总数从 3 改到 5，一批订单被处理两次、另一批没人管

现象：为了加快日终对账，运维把 shardingTotalCount 从 3 调到 5（以为“并行度更高”）。当晚对账出现两类问题：一部分订单被两个分片各处理一次（重复冲正 30 余笔，经验值需校准），另一部分订单整晚没被处理。作业日志里没有任何异常。

排查：取数路由里的 n 是 shardingTotalCount，不是在线实例数。

定位层	取证据	判定
取数 SQL	MOD(order_id, shardingTotalCount) = shardingItem	总数变了，同一个 order_id 归属就变了
分片覆盖是否完整	枚举 0..n-1 的每个分片实际处理条数	有分片为 0 条即存在覆盖缺口
实例数与分片数关系	在线实例数是否 ≥ shardingTotalCount	实例少于分片时，一个实例承担多个分片
并行度上限	真正并行度 = min(实例数, 分片总数)	实例只有 3 个时，改成 5 也不会变快
变更是否配套数据迁移	是否按新路由重跑或迁移	直接改数即出现重叠与缺口

n=3：item0 取 mod=0 的数据，item1 取 mod=1，item2 取 mod=2
n=5：mod 空间变成 0..4，原有 0/1/2 三个分片不再覆盖 3、4 这两类数据，
     而新的 0..4 分片与旧任务并发跑时，mod=0 与 mod=3 的数据可能都被同一逻辑命中

实例数变化只让分片项换宿主、不改变取数结果；分片总数变化才会改变路由。

对症做法：

1. 分片总数按“数据切分的稳定粒度”一次定足（例如固定 16 或 32），
   扩容靠加实例，不靠改总数。
2. 确实要改总数时的正确流程：
   停旧作业 → 按新路由全量重跑一次（或做数据迁移标记）→ 再启用新配置；
   绝对不允许新旧两套路由并发。
3. 每次变更前做覆盖检查：对每个分片跑一次 count 预演，
   各分片条数之和应等于全表待处理条数。
4. 处理逻辑必须幂等：分片调整期间的重叠只能靠幂等收敛，
   冲正用 (order_id, biz_date) 唯一索引。

-- 预演：确认分片覆盖不重不漏
SELECT MOD(order_id, 5) AS item, COUNT(*) FROM t_order
 WHERE status = 'PENDING' GROUP BY MOD(order_id, 5) ORDER BY item;
-- 各 item 条数之和应等于 WHERE status='PENDING' 的总数

复盘要点：判据是“分片总数变更后必须有一次全量覆盖校验”，且校验结果留档；shardingTotalCount 应视为配置里的“半不可变项”，变更要走发布评审。并行度不够时先看是不是单分片内部太慢（取数 SQL、批大小），改总数是最容易但也最容易错的那一步。

二、故障接管后从分片起点重扫，处理量翻倍

现象：某实例在跑第 8 批（共 10 批）时宿主机宕机。接管实例在几十秒后拿到该分片，但从分片起点重新扫描，把前 7 批又跑了一遍；因为处理逻辑里有一步“发对账通知”，导致 7 个批次的商户收到重复通知，客诉 20 余起（经验值，需校准）。

排查：框架接管的是“分片项的执行权”，不是“进度”。

定位层	取证据	判定
是否有 checkpoint	代码里是否把 offset 写进业务表或 offset 表	ElasticJob 无自动 checkpoint，DataflowJob 也不存 offset
接管时点	ZK 临时节点删除时间（会话过期）与新宿主开始时间	接管通常几十秒级，不是 1–3 秒
重扫范围	新宿主的首批取数条件是否含“已处理”过滤	无过滤即从头重扫
副作用幂等	通知/外部调用是否有去重键	重复通知的直接来源
流式模式	props.streaming.process 是否为 true 且处理完不更新状态	不更新状态会反复拉到同一批

会话超时方面：客户端默认请求 60s，服务端按 2×tickTime ~ 20×tickTime 协商，默认 tickTime=2000ms 时上限是 40s——所以接管延迟量级是“几十秒”，把它当秒级切换会误判故障时长。

对症做法：

// 1. 自己维护进度：每批处理完在同一事务里更新 offset，重启/接管都从这里继续
public void process(ShardingContext ctx) {
    long offset = offsetRepo.get(ctx.getJobName(), ctx.getShardingItem());    // 上次完成位置
    while (true) {
        List<Order> batch = orderRepo.fetchAfter(offset, ctx.getShardingItem(), ctx.getShardingTotal(), 500);
        if (batch.isEmpty()) break;
        txTemplate.executeWithoutResult(s -> {
            handler.handle(batch);
            offsetRepo.save(ctx.getJobName(), ctx.getShardingItem(), batch.get(batch.size() - 1).getId());
        });
        offset = batch.get(batch.size() - 1).getId();
    }
}

// 2. 副作用与状态更新分离：通知走 outbox，处理重试不会重复通知
// 3. 流式作业必须“处理完即更新业务状态”，否则 fetchData 会一直拉到同一批

# 流式处理开关：默认 false，开启后 fetchData 返回 null/空集合才停止
props.streaming.process=true

4. 接管窗口的重复只靠幂等兜住：
   每个“批”有唯一 batchKey（jobName + shardingItem + offset 区间），
   商户通知按 (batchKey, merchantId) 去重。
5. 长任务要能分段：单批 500 条而不是整表一把梭，
   重扫代价与批大小成正比，与总数据量无关。

复盘要点：判据是“实例宕机演练后，商户通知重复率应为 0”——这个演练每季度做一次，直接 kill 持有分片的实例并观察接管后的行为。设计约束要写进作业模板：ElasticJob 只保证“分片项有人跑”，不保证“跑过的位置不重跑”；进度与幂等永远是业务的责任。这一点与 XXL-JOB 的中心化调度差别最大，见 调度核心机制。

三、ZK 抖动导致作业全线暂停，被当成调度 Bug

现象：ZK 集群例行维护重启，期间所有 ElasticJob 作业停止执行约 3 分钟，告警群里报“定时任务全部失效”。维护结束后作业自动继续，没有任何实例在跑旧分片。团队最初的判断是“调度框架不可靠”。

排查：这是设计行为——失联实例主动停跑，避免与重新分片后的宿主双写。

定位层	取证据	判定
失联实例是否停跑	实例日志 Lost ZooKeeper connection, stop job	有即按设计停止
是否存在双写	ZK 不可用期间是否有两个实例跑同一分片	无即证明这个保护有效
恢复是否需干预	连接恢复后作业是否自动继续	应自动继续，不需要重启应用
暂停时长	ZK 不可用窗口 vs 作业暂停窗口	基本相等即因果成立
影响面	该作业是否属于时间敏感型（准点触发）	决定是否需要额外兜底

依赖关系要说清：ElasticJob-Lite 没有调度中心，实例既是执行器又是调度器，触发由各实例本地 cron 完成，协调与分片映射都放 ZK，所以可用性直接依赖 ZK。

对症做法：

1. 接受“ZK 不可用 → 作业暂停”这个语义，把它写进故障预案：
   ZK 维护窗口内定时任务不执行是预期行为，需要提前公告。
2. 时间敏感型作业加“错过补偿”：
   作业入口判断“上一个应触发时间是否已执行”，未执行则立即补跑一次（靠 batchKey 幂等）。
3. ZK 侧的稳定性优先：事务日志分盘、autopurge 打开、多数派余量评估，
   见 [ZK 架构与 ZAB 协议](../16_ZooKeeper/01-ZK架构与ZAB协议.md)。
4. 需要“ZK 挂了也能继续调度”的形态要换方案（中心化调度 + DB 存储，
   如 XXL-JOB），但代价是失去这里的自动分片与失联保护语义。

// 错过补偿的最小实现：以业务日期为幂等键，补跑不会重复生效
public void execute(ShardingContext ctx) {
    LocalDate due = LocalDate.now();
    if (stateRepo.isDone(ctx.getJobName(), due, ctx.getShardingItem())) return;   // 已跑过
    run(due, ctx);
    stateRepo.markDone(ctx.getJobName(), due, ctx.getShardingItem());
}

复盘要点：判据不是“作业有没有暂停”，而是“暂停期间有没有出现双写、恢复后有没有补齐”。把这两条做成演练用例：维护 ZK 期间断言无双写，恢复后断言补跑成功且不重复生效。调度系统的五个故障域（触发、分片、执行、状态、通知）里，这一条属于“协调依赖”域，见 调度系统的五个故障域。

最小上手

准备一个 ZooKeeper 3.9（单机仅用于演示，生产至少 3 节点）：

services:
  zk:
    image: zookeeper:3.9
    ports: ["2181:2181"]

Maven 坐标：org.apache.shardingsphere.elasticjob:elasticjob-lite-core:3.0.5。

public final class OrderJob implements SimpleJob {
    public void execute(ShardingContext ctx) {
        int item = ctx.getShardingItem(), total = ctx.getShardingTotalCount();
        orderMapper.scan(item, total).forEach(this::process); // 业务自己按 item/total 取数
    }
}
CoordinatorRegistryCenter reg = new ZookeeperRegistryCenter(
        new ZookeeperConfiguration("127.0.0.1:2181", "elasticjob"));
reg.init();
JobConfiguration cfg = JobConfiguration.newBuilder("orderJob", 4)
        .cron("0/30 * * * * ?")
        .shardingItemParameters("0=shanghai,1=beijing,2=guangzhou,3=shenzhen")
        .build();
new ScheduleJobBootstrap(reg, new OrderJob(), cfg).schedule();

启动后，ZK 中的实际路径是 /<namespace>/<jobName>/...，本例为 /elasticjob/orderJob/...：

/elasticjob/orderJob/
├── config                 # 作业配置 JSON
├── instances/<ip>@-@<pid> # 临时节点，在线实例
├── sharding/<item>/{instance,running,failover,misfire,disabled}
├── servers/<ip>           # 持久节点，历史服务器列表
└── leader/{election,sharding}

docker exec -it $(docker compose ps -q zk) zkCli.sh -server 127.0.0.1:2181
ls /elasticjob/orderJob/instances
get /elasticjob/orderJob/sharding/0/instance
get /elasticjob/orderJob/config
stat /elasticjob/orderJob/leader/election/instance

stat 的 ephemeralOwner 非 0 表示临时节点；实例下线后对应节点消失，sharding 下的 instance 会被重新写入。

Spring Boot 项目可用 starter elasticjob-lite-spring-boot-starter:3.0.5。脚本作业的 3.x 配置键是 elasticjob.jobs.<name>.elasticJobType: SCRIPT，脚本命令放在 props.script.command.line；脚本必须是所有节点都能访问的绝对路径，ShardingContext 的 JSON 会追加为命令参数（脚本里通常取 $1）。

生产动作

分片数按峰值实例数规划。 shardingTotalCount 至少大于峰值实例数，并让每个分片的数据量接近。具体倍数没有框架公式，按单分片耗时和扩容余量压测后确定。改小会让原先多个分片的数据映射到同一个新分片，改大则新分片没有历史进度记录，两种情况都要迁移或全量重跑。

故障转移默认关闭。 failover(false) 是默认值；打开后失败分片会被重新分配到其他实例，但不会从失败点继续，可能重复处理，必须配合幂等。monitorExecution(true) 默认打开，会写 running 节点；秒级高频短作业可以关掉，减少 ZK 写压力。

现象	先看	判断
分片反复重分配	ZK 会话是否频繁过期、instances 是否抖动	网络/GC 导致会话超时，先修连接稳定性
作业不触发	config JSON 中 disabled 是否为 true、sharding/<item>/disabled 是否存在	作业或分片被禁用，先恢复再查 cron
只有部分分片在跑	sharding/<item>/instance 是否都有值	实例数少于分片数属正常；有实例未分到项则查重新分片
多实例重复处理	取数是否用了实例数而非分片总数	应为 MOD(id, shardingTotalCount)

相关参数：maxTimeDiffSeconds 默认 -1（不检查）；reconcileIntervalMinutes 默认 10 分钟；ZK 客户端默认请求 60s 会话超时、15s 连接超时、3 次重试。

边界

• 不是数据路由器，也不是工作流引擎。 框架只分配分片号；hash、取模、范围扫描和 DAG 编排都在业务或其他系统里。
• 没有自动 checkpoint。 DataflowJob 也不保存 offset；续跑位置是业务状态，不是框架能力。
• 强依赖 ZooKeeper。 不想维护 ZK 的团队，选中心化调度的 XXL-JOB 往往更省事。
• 和消息队列的分工不同。 ElasticJob 是定时主动拉数据，MQ 是消息到了被动消费；批量对账、清理用前者，实时事件用后者。
• 不适合秒级以下高频触发。 每次执行都可能涉及 ZK 读写，高频短任务用消息队列或时间轮更合适。
• 不保证 exactly-once。 故障接管、misfire 补偿、人工重跑都可能重复执行，幂等必须自己做。

讲清楚

ElasticJob 解决的是「一批数据怎么分给多台机器并行跑」。它把一次作业切成 N 个分片项，用 ZooKeeper 选出一个 Leader，由 Leader 把分片项分给在线实例；每个实例按本地 cron 触发，只处理分给自己的项。要注意它只分配「分片号」，不分配「数据」。订单落在哪个分片，是业务代码用 hash(order_id) % shardingTotalCount 算出来的，框架不参与。所以实例扩容不会改变数据路由，改分片总数才会。

故障接管也是换宿主，不是续进度。实例挂掉后，会话过期、临时节点删除、Leader 重新分片，新实例从该分片的起点重新扫。ElasticJob 没有自动 checkpoint，DataflowJob 的 offset 也要业务自己存。因此幂等和进度表是使用它的前提，不是可选项。