XXL-JOB分片广播实战：集群定时任务并行处理与避坑指南

1. 为什么单机定时任务在集群里一定会出问题很多团队第一次把应用从单机部署改成多副本集群时都会遇到同一个尴尬场景原本在单机跑得好好的定时任务突然开始重复执行。比如每天凌晨两点给用户发结算邮件结果三台机器同时触发用户收到了三封一模一样的邮件又比如库存对账任务三个实例同时跑数据被反复覆盖最后账目全乱。这个问题的根源其实很朴素。Scheduled这类注解是绑定在 JVM 进程上的它只认自己所在的这台机器根本不知道外面还有几个兄弟实例。你部署了三个副本就等于把同一个闹钟买了三个到点当然一起响。有人会说那我加个 Redis 分布式锁不就行了锁确实能解决同一时刻只有一个实例执行的问题但它解决不了另一个更隐蔽的需求我想让三个实例一起干活每个只干一部分。举个真实例子。我做过一个批量推送项目需要给两百万用户发站内信。单机跑一轮要四十多分钟业务方嫌慢。这时候分布式锁反而成了阻碍——它强制串行三台机器只有一台在忙另外两台干看着。真正需要的是一种把任务拆开、分给所有实例并行处理的机制。这就是 XXL-JOB 分片广播模式要解决的核心问题让一次调度触发所有执行器同时给每个执行器一个编号让它们各自认领属于自己的那份数据。理解了这个出发点后面所有的配置、代码、坑点才有意义。分片广播不是更高级的定时任务它是分布式并行计算在任务调度领域的一个具体落地。你把它当成 MapReduce 里那个 Map 阶段就对了——调度中心负责切分执行器负责各自处理分片最后汇总结果。2. 分片广播的底层机制调度中心和执行器到底怎么配合2.1 一次广播调度的完整链路要搞懂分片广播得先看清楚一次调度请求从发出到执行完毕中间经过了哪些环节。XXL-JOB 的架构里有两个核心角色调度中心admin和执行器executor。普通任务的路由策略是选一台而分片广播的路由策略是全选。具体链路是这样的调度中心到点触发任务根据路由策略发现当前是SHARDING_BROADCAST于是它不会只挑一台执行器而是把注册在案的所有执行器地址全部拉出来逐个发起调度请求。注意这里是逐个发起不是发一条消息让执行器自己抢。每个执行器收到的请求里都带着两个关键参数分片总数shardTotal和当前分片序号shardIndex。执行器收到请求后把这两个参数塞进任务方法的入参里。你的业务代码通过XxlJobHelper.getShardIndex()和XxlJobHelper.getShardTotal()就能拿到。分片序号从 0 开始比如三台机器序号就是 0、1、2。业务代码拿到序号后用取模的方式决定我该处理哪些数据。这里有个容易被忽略的细节分片序号和执行器实例不是永久绑定的。今天序号 0 可能是 A 机器明天 A 机器下线了序号 0 就落到 B 机器头上。所以你的业务逻辑绝对不能依赖序号 0 一定是某台特定机器这种假设只能依赖序号 0 处理 id 取模等于 0 的数据这种纯计算逻辑。2.2 分片参数是怎么传进业务方法的很多人第一次写分片任务会习惯性地在方法签名里加参数结果发现拿不到值。XXL-JOB 的新版本2.2.0 之后推荐用XxlJobHelper这个工具类来获取上下文而不是靠方法参数注入。XxlJob(shardingJobHandler) public void shardingJobHandler() throws Exception { // 获取分片序号和总分片数 int shardIndex XxlJobHelper.getShardIndex(); int shardTotal XxlJobHelper.getShardTotal(); XxlJobHelper.log(当前分片序号{}分片总数{}, shardIndex, shardTotal); // 业务逻辑只处理属于自己分片的数据 ListLong userIds userMapper.selectByShard(shardIndex, shardTotal); for (Long userId : userIds) { processUser(userId); } }对应的 SQL 通常长这样SELECT id, name FROM user WHERE MOD(id, #{shardTotal}) #{shardIndex} LIMIT 1000用MOD(id, shardTotal) shardIndex这个条件就能保证每条数据只会被一个分片捞到不会重复也不会遗漏。这是分片广播最经典的用法也是面试里被问得最多的点。2.3 分片总数到底等于几这是新手最容易踩的坑。分片总数不等于你配置的机器数量而是等于当前在线执行器实例的数量。调度中心在发起广播前会实时查询执行器注册表拿到当前活着的实例列表然后shardTotal就等于这个列表的长度。这意味着什么意味着你的分片数是动态的。早上三台机器在线shardTotal就是 3中午扩容到五台shardTotal就变成 5。业务代码必须能适应这种变化不能写死。我见过有人把分片逻辑写成if (shardIndex 0) { 处理前半部分 } else { 处理后半部分 }结果扩容到三台后直接崩了——第三台机器不知道该干啥。正确的做法永远是用取模运算动态计算让分片数和数据切分规则解耦。这样无论实例怎么增减数据都能被正确覆盖。3. 从零搭一个分片广播任务配置、代码、验证3.1 调度中心和执行器的版本对齐动手之前先确认版本。XXL-JOB 在 2.1.0 之后对分片广播的支持才比较完善2.2.0 引入了XxlJobHelper2.3.0 之后对注册中心做了优化。我建议直接用 2.3.x 或 2.4.x老版本在实例上下线时容易出现分片数计算不准的问题。调度中心xxl-job-admin和执行器xxl-job-core的版本必须一致这是硬性要求。我踩过一次坑admin 用的 2.3.0执行器依赖写成了 2.2.0结果分片参数传过去是 null排查了大半天才发现是版本不匹配。Maven 里这样对齐dependency groupIdcom.xuxueli/groupId artifactIdxxl-job-core/artifactId version2.4.0/version /dependency3.2 执行器配置文件的关键项执行器的application.properties里和分片广播相关的配置其实不多但每一项都不能错# 调度中心地址多个用逗号分隔 xxl.job.admin.addresseshttp://127.0.0.1:8080/xxl-job-admin # 执行器注册方式推荐自动注册 xxl.job.executor.address xxl.job.executor.ip xxl.job.executor.port9999 xxl.job.executor.logpath/data/applogs/xxl-job/jobhandler xxl.job.executor.logretentiondays30 # 执行器AppName调度中心靠这个找到执行器 xxl.job.executor.appnamexxl-job-executor-sharding重点说appname。调度中心在配置任务时要选择执行器这个执行器就是靠appname关联的。如果你部署了三个实例它们的appname必须完全相同这样调度中心才会认为它们是同一个执行器下的三个节点广播时才会把三个都算进去。如果appname写得不一致调度中心会当成三个独立的执行器分片逻辑就乱了。port这一项在多实例部署时要注意如果三台机器在不同主机上端口可以都用 9999如果在同一台机器上跑多个实例本地测试常见端口必须错开否则后启动的会绑定失败。3.3 调度中心里怎么配这个任务登录调度中心新建任务几个关键配置项配置项值说明路由策略分片广播核心选错就不是广播了运行模式BEAN用注解方式注册的处理器JobHandlershardingJobHandler和代码里XxlJob的值对应阻塞处理策略丢弃后续调度分片任务通常不希望堆积调度过期策略忽略避免补跑导致数据重复路由策略选分片广播是整件事的开关。选成轮询或第一个任务只会在一台机器上跑分片参数虽然也会传但shardTotal永远是 1等于没分片。阻塞处理策略我一般选丢弃后续调度。因为分片任务往往是批处理一轮没跑完又来一轮容易造成数据竞争。如果你的任务必须串行那就得靠业务层面的幂等来兜底。3.4 本地模拟多实例验证本地验证分片广播最省事的办法是启动多个执行器实例改端口和日志路径java -jar executor.jar --server.port8081 --xxl.job.executor.port9991 java -jar executor.jar --server.port8082 --xxl.job.executor.port9992 java -jar executor.jar --server.port8083 --xxl.job.executor.port9993三个实例的appname保持一致。启动后去调度中心的执行器管理页面应该能看到这个执行器下面挂着三个在线节点。这时候手动触发一次任务看日志三个实例的日志里应该分别打印出shardIndex0/1/2shardTotal3。如果只看到一个实例在跑八成是appname不一致或者路由策略没选对。提示本地测试时如果发现分片数不对先检查执行器注册列表里到底有几个在线节点。调度中心的分片数是实时算的注册表不准分片就一定不准。4. 分片逻辑写不对等于白配4.1 取模分片的正确姿势分片广播最容易出问题的地方不在配置而在业务代码里的数据切分逻辑。取模是最常用的方式但写法有讲究。// 推荐用分片总数和序号做取模 int shardIndex XxlJobHelper.getShardIndex(); int shardTotal XxlJobHelper.getShardTotal(); // 查询属于当前分片的数据 ListTask tasks taskMapper.selectSharding(shardIndex, shardTotal);对应的 MyBatis 映射select idselectSharding resultTypeTask SELECT * FROM task WHERE status 0 AND MOD(id, #{shardTotal}) #{shardIndex} ORDER BY id LIMIT #{pageSize} /select这里有个性能陷阱MOD(id, shardTotal)在数据量大时会导致全表扫描因为函数作用在列上索引失效。如果表有几百万行这个查询会非常慢。优化思路有两个一是用id的范围分片代替取模二是提前算好分片字段并建索引。范围分片的写法SELECT * FROM task WHERE status 0 AND id % #{shardTotal} #{shardIndex} AND id #{lastMaxId} ORDER BY id LIMIT #{pageSize}配合游标lastMaxId翻页既能走索引又能避免深分页。这是我在实际项目里验证过的方案两百万数据的分片查询从十几秒降到几百毫秒。4.2 分片数变化时的数据一致性前面说过分片数是动态的。三台机器时MOD(id, 3)扩容到五台变成MOD(id, 5)。如果任务跑到一半扩容了会发生什么假设第一轮用 3 个分片处理了 id 1 到 1000第二轮扩容到 5 个分片MOD(id, 5)的切分方式和MOD(id, 3)完全不同。原本 id3 的数据在第一轮属于分片 0第二轮可能属于分片 3。如果任务没有幂等保护这条数据会被处理两次。解决办法有两个层面。任务层面给每条数据的处理加上状态标记处理过的打上processed1查询时过滤掉。调度层面尽量避免在任务执行期间扩容或者把分片任务设计成每次全量重算而不是增量累加。我个人的经验是批处理类任务用状态标记最稳妥因为扩容是运维的常规操作你没法保证它不在任务执行时发生。4.3 分片任务里的日志和监控分片任务出问题时排查比单机任务麻烦得多因为日志散在多个实例上。XXL-JOB 提供了XxlJobHelper.log()方法它会把日志回传到调度中心在调度日志页面能看到每个分片的执行情况。XxlJobHelper.log(分片 {} 开始处理待处理数量{}, shardIndex, tasks.size()); // ... 处理逻辑 XxlJobHelper.log(分片 {} 处理完成成功{}失败{}, shardIndex, success, fail);这个日志是分片任务排查的生命线。我习惯在每个分片的开头和结尾都打一条这样在调度中心一眼就能看出哪个分片没跑、哪个分片卡住了。如果某个分片的日志一直不出现说明那台执行器可能掉线了或者任务分发时没收到请求。另外XxlJobHelper.log()的日志有长度限制默认单条日志不能太长。如果要打大量内容建议只打摘要详细日志写到本地文件。5. 那些文档里不会写的坑5.1 执行器掉线导致的分片空洞这是分片广播最隐蔽的问题。假设三台机器分片 0、1、2。任务执行到一半分片 1 的机器突然挂了。会发生什么调度中心在发起调度时分片数是基于调度那一刻的在线实例算的。如果机器是在任务执行过程中挂的那分片 1 的数据就没人处理了而且调度中心不会自动把分片 1 重新分配给其他机器。结果就是这部分数据被漏掉直到下一次调度才会被重新捞起来。如果任务对实时性要求高这个空洞是不能接受的。应对方案是补偿机制任务执行完后检查每个分片是否都上报了完成状态没上报的分片由调度中心或某个协调者重新触发。XXL-JOB 本身不提供这个能力需要自己在业务层实现。我通常会在任务表里加一个shard_status字段每个分片完成后写入自己的状态然后有一个独立的巡检任务定期检查有没有超时未完成的分片。5.2 分片序号为负数的诡异情况有次线上排查发现日志里打印出shardIndex-1。查了半天原因是执行器版本和调度中心版本不一致老版本执行器在拿不到分片参数时默认返回 -1。这种问题不会报错只会让业务逻辑静默地处理错误的数据范围。防御性写法int shardIndex XxlJobHelper.getShardIndex(); int shardTotal XxlJobHelper.getShardTotal(); if (shardIndex 0 || shardTotal 0 || shardIndex shardTotal) { XxlJobHelper.log(分片参数异常shardIndex{}, shardTotal{}, shardIndex, shardTotal); return; }这段校验看起来多余但能帮你快速定位版本不匹配的问题。我现在的习惯是每个分片任务开头都加这段成本很低收益很高。5.3 分片任务和事务的冲突分片任务通常是批处理一批处理几百上千条。如果整个方法包在一个大事务里一旦某条数据出错整批回滚前面处理成功的也白干了。而且大事务会长时间占用数据库连接分片数一多连接池直接被打满。正确的做法是分批提交每处理 N 条提交一次事务。可以用编程式事务int batchSize 100; ListTask tasks taskMapper.selectSharding(shardIndex, shardTotal); for (int i 0; i tasks.size(); i batchSize) { ListTask batch tasks.subList(i, Math.min(i batchSize, tasks.size())); transactionTemplate.execute(status - { for (Task task : batch) { processTask(task); } return null; }); }这样单批失败只影响这一批前面的成果保留。配合状态标记下一轮调度可以继续处理失败的批次。5.4 分片数远大于数据量时的空转如果数据只有 100 条但你有 50 台机器分片数就是 50。大部分分片查出来是空的白白浪费调度资源。这种情况在小数据量、大集群的场景下很常见。优化思路是限制分片数的上限。可以在业务代码里判断如果数据量小于某个阈值就只让分片 0 处理全部数据long totalCount taskMapper.countPending(); if (totalCount 1000) { // 数据量小只让分片 0 处理 if (shardIndex ! 0) { return; } // 分片 0 处理全部 processAll(); } else { // 正常分片处理 processSharding(shardIndex, shardTotal); }这个判断逻辑让分片机制在小数据量时退化成单机执行避免无谓的空转。阈值设多少取决于你的单条处理耗时和集群规模我一般设在 1000 到 5000 之间。6. 分片广播和其他方案的取舍6.1 和消息队列削峰的对比有人会问批量处理为什么不用消息队列把两百万用户丢进 MQ多个消费者并行消费不也能达到并行处理的效果吗两者确实有重叠但适用场景不同。MQ 适合流式、实时、无状态的处理消息发出去就不管了消费失败靠重试机制。分片广播适合批量、定时、有状态的处理比如每天凌晨对账、每月生成报表。这类任务需要知道总共处理了多少、成功多少、失败多少需要能重新触发整个批次这些是 MQ 不擅长的。我的经验是如果是用户触发的实时任务用 MQ如果是定时批处理用分片广播。两者不是替代关系很多系统里是共存的。6.2 和 ElasticJob 的分片对比ElasticJob 也支持分片而且它的分片是持久化的——分片信息存在注册中心实例上下线时会自动重新分片。XXL-JOB 的分片是每次调度时实时计算的不持久化。这个差异导致 ElasticJob 在实例频繁上下线时更稳定因为它有重新分片的机制。但 ElasticJob 的运维复杂度更高需要额外的注册中心ZooKeeper 或 Nacos。XXL-JOB 胜在轻量调度中心自带注册功能部署简单。选型建议如果你的集群规模不大十几台以内实例变动不频繁XXL-JOB 足够用。如果集群规模大、弹性伸缩频繁ElasticJob 的持久化分片更合适。我做过一个上百实例的项目最后选了 ElasticJob就是因为 XXL-JOB 在实例频繁上下线时分片数抖动太厉害。6.3 分片广播的适用边界不是所有任务都适合分片广播。判断标准很简单任务能否被拆成互不依赖的独立子任务。能拆就用分片不能拆就老老实实用单机加锁。比如生成全站统计报表这种任务它需要汇总所有数据拆开反而要合并结果用分片就是自找麻烦。而给每个用户发通知这种任务用户之间互不依赖天然适合分片。还有一个边界是数据倾斜。如果数据分布不均匀比如 id 取模后某个分片的数据量是其他分片的十倍那这个分片就会成为瓶颈其他分片早早跑完干等着。解决办法是换分片键用更均匀的字段比如用户 id 的哈希值代替自增 id。这个坑我在一个订单处理项目里踩过订单 id 是按时间递增的取模后最近的分片数据量爆炸后来改成按用户 id 哈希才解决。7. 几个实战中的参数调优经验7.1 调度超时时间怎么设调度中心里有个任务超时时间配置默认是 0不限制。分片任务一定要设这个值否则某个分片卡死整个任务永远不结束。设多少合适取决于你的单批处理耗时。我的经验公式是超时时间 单批处理耗时 × 3 30秒。留三倍余量是为了应对数据量波动加 30 秒是给网络和调度留缓冲。比如单批处理 2 分钟超时设 7 分钟左右。超时后 XXL-JOB 会中断任务但注意它只是标记任务失败不会真的 kill 掉执行线程。所以业务代码里要有响应中断的逻辑比如在循环里检查Thread.currentThread().isInterrupted()。7.2 失败重试的坑XXL-JOB 支持失败重试配置项是失败重试次数。分片任务开重试要谨慎因为重试是整个任务重试不是只重试失败的分片。如果三个分片里只有一个失败重试会把三个分片全部重跑一遍已经成功的分片会重复处理。如果任务不是幂等的这个重试就是灾难。我的建议是分片任务默认关闭重试靠业务层的状态标记和补偿任务来处理失败。如果一定要开重试确保业务逻辑幂等。7.3 日志保留天数的权衡xxl.job.executor.logretentiondays控制执行器本地日志的保留天数默认 30 天。分片任务日志量大如果集群规模大30 天可能占满磁盘。我一般设成 7 天因为调度中心已经存了关键日志本地日志主要用于详细排查7 天足够覆盖大部分问题。调度中心的日志也有清理策略在 admin 的配置文件里默认保留 30 天。这个可以按需调整但别设太短否则出了问题想查历史日志都查不到。8. 一个完整的分片任务模板把前面所有经验揉在一起给一个可以直接抄的模板Component public class ShardingJobHandler { Autowired private TaskMapper taskMapper; Autowired private TransactionTemplate transactionTemplate; private static final int BATCH_SIZE 100; private static final long SMALL_DATA_THRESHOLD 1000; XxlJob(shardingJobHandler) public void execute() { int shardIndex XxlJobHelper.getShardIndex(); int shardTotal XxlJobHelper.getShardTotal(); // 参数校验 if (shardIndex 0 || shardTotal 0 || shardIndex shardTotal) { XxlJobHelper.log(分片参数异常index{}, total{}, shardIndex, shardTotal); XxlJobHelper.handleFail(分片参数异常); return; } // 小数据量退化为单机 long totalCount taskMapper.countPending(); if (totalCount SMALL_DATA_THRESHOLD shardIndex ! 0) { XxlJobHelper.log(数据量小分片 {} 跳过, shardIndex); return; } XxlJobHelper.log(分片 {} 开始总数 {}待处理 {}, shardIndex, shardTotal, totalCount); long lastMaxId 0; int successCount 0; int failCount 0; while (true) { // 游标分页避免深分页 ListTask tasks taskMapper.selectShardingByCursor( shardIndex, shardTotal, lastMaxId, BATCH_SIZE); if (tasks.isEmpty()) { break; } for (Task task : tasks) { try { transactionTemplate.execute(status - { processTask(task); return null; }); successCount; } catch (Exception e) { failCount; XxlJobHelper.log(处理失败id{}, error{}, task.getId(), e.getMessage()); } lastMaxId Math.max(lastMaxId, task.getId()); } } XxlJobHelper.log(分片 {} 完成成功 {}失败 {}, shardIndex, successCount, failCount); if (failCount 0) { XxlJobHelper.handleFail(存在失败记录 failCount); } } private void processTask(Task task) { // 业务处理逻辑 } }这个模板覆盖了参数校验、小数据退化、游标分页、分批事务、失败统计几个关键点。你可以根据自己的业务替换processTask和查询 SQL其余部分基本不用改。9. 排查分片问题的固定套路分片任务出问题排查顺序我总结成一套固定流程照着走基本能定位第一步看调度中心的调度日志。确认任务是否触发了、触发了几个分片、每个分片的执行状态。如果只触发了一个分片问题在路由策略或执行器注册。第二步看执行器注册列表。确认在线实例数和预期是否一致。实例数不对分片数就不对。第三步看每个分片的业务日志。通过XxlJobHelper.log()回传的日志确认每个分片处理的数据范围。如果某个分片处理的数据为空检查取模逻辑。第四步看数据是否重复或遗漏。用 SQL 统计每个分片处理的数据量加起来是否等于总数。不等就说明分片逻辑有问题。第五步看是否有分片空洞。检查是否有分片没有上报完成状态这通常意味着执行器掉线或任务超时。这套流程我在多个项目里用过大部分分片问题都能在前三步定位。第四步和第五步主要用于排查数据一致性问题需要结合业务表的状态字段来分析。分片广播这个机制配置本身不复杂难的是业务逻辑的正确性和边界情况的处理。把取模逻辑写对、把分片数变化考虑进去、把失败补偿做好基本就能稳定运行。剩下的就是根据实际数据量和集群规模调参数这部分没有标准答案得靠实际跑几轮来摸索。