1. 先搞清楚Dataflow作业到底解决什么问题聊到 Elastic-Job很多人第一反应就是写一个SimpleJob到点执行一个方法跑完收工。但实际做业务系统时你会发现纯粹按 cron 触发方法根本不够用。比如订单超时未支付要批量关闭、支付回调丢了要定时补齐状态、日志表里几十万条脏数据要分批修正。这类场景有个共同点既要“按条件拉数据”又要“处理这批数据”而且数据量可能很大一次取不完、也不能一把梭全量处理。Dataflow 作业就是 Elastic-Job 专门为这类场景提供的作业模型。它把一次作业拆成两个阶段fetchData负责按分片条件捞出一批待处理数据processData负责处理这批数据。框架负责调度、分片、生命周期业务只关心“取什么”和“怎么处理”。数据按批拉取天然适合大数量、增量、可断点续跑的业务。这篇文章适合两种人一是你正从 Simple 作业往 Dataflow 作业过渡想知道接口怎么写、参数怎么配二是你已经在生产用了 Dataflow遇到了像“死循环”“数据漏处理”这样说不清道不明的诡异问题想找排查思路。两种情况我都会覆盖把接口语义、参数原理、代码示例、踩坑记录揉在一起讲清楚。1.1 从Simple作业的瓶颈说起先别急着写代码我们想清楚 Dataflow 到底补上了 Simple 的什么短板。Simple 作业就是一个方法cron 到点执行比如public class OrderCloseSimpleJob implements SimpleJob { Override public void execute(ShardingContext context) { // 查超时订单 ListOrderEntity orders orderMapper.selectTimeoutOrders(); // 批量关闭 for (OrderEntity order : orders) { orderMapper.markClosed(order.getId()); } } }这段代码单机跑没问题但一旦上了生产问题就来了。第一数据量大了以后全表扫描一次取出来再循环处理内存和数据库连接都扛不住。第二多实例部署时如果没做控制同一个任务每个实例都会执行一遍数据会被重复处理。第三处理过程中某条数据失败导致整个方法中断下一轮 cron 又从头开始没有“断点续跑”的概念。第四也是我感受最深的取数和处理逻辑揉在一个方法里出了线上问题你很难判断到底是“取漏了”还是“处理错了”。Simple 作业适合“到点干一件事”的简单场景比如清理临时目录、生成日报表、发送提醒通知。但一旦这个事里面出现了“从某张表拉一批待处理数据然后逐个更新状态”这样的循环流程你就应该考虑 Dataflow 了。1.2 拆分“取数”与“处理”框架来兜底Dataflow 的核心思想是把数据驱动型任务拆成两个阶段并且由框架保证每个阶段的行为可控框架在 cron 到点后调用你的fetchData(shardingContext)你返回一批要处理的数据如果返回的数据不为空框架接着调用processData(shardingContext, data)处理这批数据根据streamingProcess配置决定这一轮是“取一次处理一次就结束”还是“取完一批处理完继续取下一批直到取不到为止”。你可以把它想象成一条生产流水线fetchData是“从货架取一箱货”processData是“拆箱质检入库”。货架空了就停下等下次定时触发开了流式模式后取完一箱立刻取下一箱一直做到货架空了为止。这个拆分最大的价值在于可测试性。取数逻辑可以单独验证处理逻辑也可以单独用虚构数据跑测试。线上出问题时通过日志能清楚看到是“拉不出来数据”还是“拉了数据处理失败”排查范围一下子缩小了一半。从这个角度说Dataflow 不是给 Simple 的简单升级它改变的是你对任务建模的思路数据本身才是任务的核心时间触发只是启动器。1.3 三种作业类型怎么选Elastic-Job 提供了 Simple、Dataflow、Script 三种作业类型不少人配置的时候不知道选哪种这里给一个对照作业类型触发模型最典型的使用场景一句话选型建议Simple到点执行一个方法报表生成、定时清理、发送通知和“数据量”无关干完就走Dataflow拉一批数据并处理状态补偿、批量流转、积压数据处理有大量数据要循环消化Script执行 shell / python 脚本跨语言任务、运维脚本逻辑不在 Java 里时使用你完全可以手动在 Simple 里写 while 循环把 Dataflow 的活儿干了但那样的话分片上下文、流式控制、失败重试、作业监控这些能力就全都要自己造轮子。框架给不了你业务智能但它能给你稳定的调度骨架这就是用 Dataflow 而不是手写循环的理由。2. Dataflow接口与核心参数解析2.1 接口长什么样2.x 与 3.x 对照先说接口本身。Elastic-Job 2.x 的 DataflowJob 是这样的import com.dangdang.ddframe.job.api.ShardingContext; import com.dangdang.ddframe.job.api.dataflow.DataflowJob; public class OrderCloseDataflowJob implements DataflowJobOrderEntity { Override public ListOrderEntity fetchData(ShardingContext shardingContext) { // 按分片条件捞一批待处理数据 return null; } Override public boolean processData(ShardingContext shardingContext, ListOrderEntity data) { // 处理数据返回是否继续拉取下一批 return true; } }Elastic-Job 3.x 的接口变成这样import org.apache.shardingsphere.elasticjob.api.ShardingContext; import org.apache.shardingsphere.elasticjob.api.dataflow.DataflowJob; public class OrderCloseDataflowJob implements DataflowJobOrderEntity { Override public ListOrderEntity fetchData(ShardingContext shardingContext) { // 按分片条件捞一批待处理数据 return null; } Override public void processData(ShardingContext shardingContext, ListOrderEntity data) { // 处理数据异常即失败 } }两个版本之间最大的变化是processData的返回值从boolean变成了void。在老版本里返回false代表“本轮不继续拉取下一批”在新版本里这个控制逻辑被去掉了失败统一通过抛异常表达。后面讲踩坑的时候我会再强调这点。2.2 fetchData 和 processData 各该干什么fetchData是数据入口写它的时候有几个原则返回null或空列表表示“当前没有要处理的数据”这一轮调度结束一次取多少条很关键。我一般控制在 100 到 500 条之间取决于单条处理的耗时和数据库连接资源。取太少框架频繁调用开销大取太多内存占用和单个事务时间都会失控必须用shardingContext.getShardingItem()过滤数据。这个值表示当前作业实例负责的分片项比如总分片数为 4实例 A 可能拿到 0 和 2实例 B 拿到 1 和 3。你不按它过滤多实例部署时数据必然重复处理一个很容易被忽略的坑取出来的数据处理之后状态必须发生变化否则在流式模式下会反复拉取同一批数据形成死循环。这一点放到第 4 节展开。processData是数据处理入口写的时候注意尽量保证幂等。作业失败后会重跑processData可能被重复调用所以处理逻辑要能接受“这条数据已处理过”的情况不要在整个列表外面套一个大事务。建议每条数据一个短事务或者按小批次提交。这样单条失败不会把整批回滚下一轮还能只补未处理成功的部分3.x 版本中方法内抛异常就代表处理失败框架会中止当前轮次下个 cron 周期再进来。2.3 streaming-process一次任务到底跑几轮这是 Dataflow 作业最重要的开关也是新手最容易搞混的配置项。streaming-processfalse默认值时每次 cron 触发后框架只执行一次fetchDataprocessData不管取到多少数据处理完这一批就结束等下一个调度周期。适合数据量不大、单批处理时间可控的常规场景。streaming-processtrue时框架进入流式循环取一批 → 处理一批 → 再取一批 → 再处理直到fetchData返回空列表才结束本次触发。适合需要“追积压”的场景比如每分钟补偿一次每次都尽可能把当前积压的清空。一个常见误区是认为 streaming 模式性能更好。其实它只是改变循环控制方式并不改变你单批处理的速度。它的价值在于一次触发内能连续消化多批数据而不是每批都要等下一个 cron 周期这对积压追赶非常有效。我在生产上踩过一次坑明明只是普通的定时状态扫描数据量也不大却把streaming-process设成了true线上任务偶尔会出现“一次触发跑了几分钟”的现象表面看是性能问题其实是流式循环把本应分摊到多个周期的活集中在一次做完了。所以我的建议是默认先开 false确有必要追数据积压时再开 true。2.4 ShardingContext 怎么参与数据路由ShardingContext 是 Dataflow 作业里绕不开的对象它提供了几个关键信息方法含义getShardingItem()当前实例被分配到的分片项编号从 0 开始getShardingTotalCount()总分片数getShardingParameter()当前分片项对应的业务参数来自shardingItemParameters配置getJobParameter()作业级参数所有分片共享最常规的用法是配合数据库取模来路由。比如总分片数为 4查询条件里带上MOD(id, 4) shardingItem这样每个实例只会处理自己那 1/4 的数据天然避免重复。但我要提醒一点MOD(id, n)这种方式在数据量小的时候没问题一旦表数据量大、id 又不连续删除、合并、横向拆分都会导致空洞模运算容易造成分片倾斜。更可靠的方式是用主键区间切分或者用专门的业务路由字段后面排查章节我会给出具体做法。3. 实操实现一个订单超时补偿 Dataflow 作业3.1 场景与表结构设计为了把流程完整串起来我设计一个模拟项目信号订单超时未支付自动关闭。订单表结构如下CREATE TABLE t_order ( id bigint NOT NULL AUTO_INCREMENT, order_no varchar(64) NOT NULL, status int NOT NULL DEFAULT 0 COMMENT 0待支付 1已支付 2已关闭, create_time datetime NOT NULL, close_time datetime DEFAULT NULL, PRIMARY KEY (id), KEY idx_status_create (status, create_time) ) ENGINE InnoDB COMMENT 订单表;业务规则很简单订单创建 30 分钟内未支付由作业扫描并标记为关闭。这个场景非常适合 Dataflow因为可能每天有大量订单进入“超时待处理”状态需要分批拉取、逐条更新、持续消化。为了支持分片我给 Mapper 加一个按模运算过滤的查询方法。注意 SQL 里要带LIMIT这是防止一次拉取过多数据的关键。Mapper public interface OrderMapper { ListOrderEntity selectTimeoutOrders(Param(shardingItem) int shardingItem, Param(shardingTotalCount) int shardingTotalCount, Param(limit) int limit); int markClosed(Param(id) long id, Param(closeTime) LocalDateTime closeTime); }对应的 XML 查询语句select idselectTimeoutOrders resultTypeOrderEntity SELECT id, order_no, status, pay_status, create_time FROM t_order WHERE status 0 AND MOD(id, #{shardingTotalCount}) #{shardingItem} AND create_time lt; DATE_SUB(NOW(), INTERVAL 30 MINUTE) ORDER BY create_time ASC LIMIT #{limit} /select3.2 编写 DataflowJob 实现类作业实现类代码如下2.x 版本public class OrderCloseDataflowJob implements DataflowJobOrderEntity { private static final int BATCH_SIZE 200; private final OrderMapper orderMapper; public OrderCloseDataflowJob(OrderMapper orderMapper) { this.orderMapper orderMapper; } Override public ListOrderEntity fetchData(ShardingContext shardingContext) { return orderMapper.selectTimeoutOrders( shardingContext.getShardingItem(), shardingContext.getShardingTotalCount(), BATCH_SIZE); } Override public boolean processData(ShardingContext shardingContext, ListOrderEntity data) { for (OrderEntity order : data) { orderMapper.markClosed(order.getId(), LocalDateTime.now()); } return true; } }3.x 版本只需要把processData的返回值改成voidOverride public void processData(ShardingContext shardingContext, ListOrderEntity data) { for (OrderEntity order : data) { orderMapper.markClosed(order.getId(), LocalDateTime.now()); } }这里我最想强调的一点markClosed必须把status从 0 改成 2。因为fetchData的查询条件就是status 0只要状态更新了这条数据下一轮就不会再被捞出来。这是 Dataflow 作业不出死循环的根基。3.3 注册作业的三种配置方式先看 3.x 的 Java API 方式。这是新项目最推荐的写法配置清晰、不依赖 SpringZookeeperConfiguration zkConfig new ZookeeperConfiguration(localhost:2181, elastic-job-demo); CoordinatorRegistryCenter regCenter new ZookeeperRegistryCenter(zkConfig); regCenter.init(); JobConfiguration jobConfig JobConfiguration.newBuilder(orderCloseJob, 4) .cron(0 */1 * * * ?) .streamingProcess(true) .build(); new ScheduleJobBootstrap(regCenter, new OrderCloseDataflowJob(orderMapper), jobConfig).schedule();如果用 Spring Boot Starter在application.yml里这样配elasticjob: reg-center: server-lists: localhost:2181 namespace: elastic-job-demo jobs: orderCloseJob: elastic-job-class: com.example.job.OrderCloseDataflowJob cron: 0 */1 * * * ? sharding-total-count: 4 streaming-process: true老项目还在用 2.x 的话XML 方式最常见job:dataflow idorderCloseJob classcom.example.job.OrderCloseDataflowJob registry-center-refregCenter cron0 */1 * * * ? sharding-total-count4 streaming-processtrue /2.x 的配置项里streaming-process必须和作业类型配套。如果配job:simple却带了这个参数虽然不会报错但参数不生效容易造成“配了不生效”的错觉。提示3.x 已经不再推荐老式 XML 命名空间方式官方主推 Spring Boot Starter 和 Java API。迁移时别再把老配置直接搬过来。3.4 启动观察日志、分片与数据变化配置好了以后启动两个应用实例做一次完整的验证。第一步看分片分配。日志里会打印每个实例负责的分片项正常情况下一个实例拿到[0, 2]另一个拿到[1, 3]这说明 ZK 注册和分片协商正常。第二步造测试数据。往t_order里插入几条 40 分钟前创建、状态为 0 的订单等 cron 触发后查询数据库这几条订单的status应该变成 2close_time有值。第三步观察流式循环。配置了streaming-processtrue时日志里会连续出现多轮fetch size200、fetch size100最后出现一次fetch empty表示本次触发结束。如果只看到一轮 fetch 就结束了要么是数据不够要么是配置没生效可以去注册中心确认作业属性。第四步验证多实例不重复。两个实例同时运行时同一订单只会被其中一个实例关闭。判断标准MOD(id, 4)的结果只属于一个实例处理日志中不会出现两条相同的订单 id。4. 常见问题与排查技巧实录4.1 streaming 模式下被“死循环”卡住这是 Dataflow 作业最经典的事故。表现是作业一直在跑日志里不断打印 fetch 到数据、处理数据数据库里某些记录反复出现作业线程被占住不放。我当时的排查思路是死循环的本质一定是“处理完的数据仍然满足 fetchData 的查询条件”。具体原因通常有几种processData只做了逻辑判断没有更新查询条件里的状态字段更新的是另一张关联表而 fetchData 查的是主表更新语句的 WHERE 条件和查询语句不一致导致某些记录永远改不动处理抛了异常被吞掉状态没更新但循环还在继续。排查步骤我整理成速查表步骤操作目的1把streaming-process临时改成 false确认问题是否由流式循环触发2在 fetchData 里打印返回数据的 id 集合判断是不是同一批数据反复出现3对比 fetchData 的 WHERE 和 processData 的 UPDATE 条件找出状态永远不变的数据4在 processData 里加异常日志不要吞异常确认是否有静默失败一个实用的兜底方案在fetchData的 SQL 里加一个处理次数限制比如额外加一个retry_count 5的条件避免单条脏数据把整个作业拖死。4.2 分片不均导致的数据漏处理用MOD(id, shardingTotalCount)做路由最常见的问题是 id 不连续导致分片倾斜。比如订单表里大量订单是通过中间表导入的id 分布不均结果某个分片的数据特别多另一个分片几乎没有数据表现为“某些实例忙死某些实例闲死”。如果是漏处理问题更隐蔽你以为 4 个分片覆盖了全表但MOD的分布和实际数据分布不一致某个值得处理的数据恰好落在没有数据的分片里。排查的方法是分别把shardingItem0到shardingItem3的查询结果数量统计出来看是否符合预期。对于数据量大的表我更推荐用主键区间切分而不是模运算。思路是先用一条 SQL 查出MIN(id)和MAX(id)再按总分片数均分成几个区间每个实例处理自己区间内的id这样即使 id 有空洞区间边界也不会漏数据。但要记住区间切分在数据持续增长时头部和尾部区间天然不均衡需要根据业务特点取舍。4.3 processData 抛异常后的状态一致性Dataflow 的处理语义是“至少处理一次”不是“精确一次”。也就是说某批数据处理到一半抛了异常已经成功的部分是不会回滚的未成功的部分下一轮会重新被捞出来。所以你在写 processData 时必须接受这个现实然后用幂等设计来兜底。我自己的习惯是每条数据单独调用一个更新方法不用大事务包住整个 List更新语句里带上条件判断比如UPDATE t_order SET status 2 WHERE id ? AND status 0这样重复处理不会覆盖新状态如果 processData 里要调用外部接口发短信、调支付查询一定要有去重机制因为作业重试会重新执行整批数据业务异常要分类可重试的比如下游暂时不可用就让作业下轮重跑不可重试的比如参数非法要记录到异常表或日志里避免永远卡在队列中。如果你还在用 2.x特别提醒一下processData返回false时框架会停止本轮流式循环不再拉取下一批数据。很多人误把false当“处理失败”的标记其实它更多是流式循环的停止开关。处理失败应该抛异常而不是静默返回 false否则数据会在数据库里原地不动而作业看起来一切正常。4.4 从 2.x 迁移到 3.x 的坑我迁移过一个跑了一年的老作业过程没有想象中顺畅。主要坑点归纳如下第一个是包名和类名变化。原来 import 的是com.dangdang.ddframe.job.api.dataflow.DataflowJob3.x 改成了org.apache.shardingsphere.elasticjob.api.dataflow.DataflowJobIDE 直接全红只能批量替换。第二个是processData返回值变化。刚才提到过2.x 返回 boolean3.x 返回 void。如果你的老代码里维护了一个return false的分支迁移时千万不要直接删掉或改成return要先想清楚这个分支原来想表达什么改成抛出业务异常或者记录日志。第三个是作业启动方式变化。2.x 用JobSchedulerLiteJobConfiguration3.x 用ScheduleJobBootstrap定时作业和OneOffJobBootstrap一次性作业。对于 Dataflow 这种持续型作业用ScheduleJobBootstrap就行。第四个是 Spring 配置方式变化。2.x 的 XML namespace 在 3.x 不再推荐官方主推注解和 Java API。如果你项目里大量用了 XML 配置迁移前先评估改动成本。4.5 性能调优的几个参数Dataflow 作业的性能瓶颈不在框架而在你 fetchData 的 SQL 和生产侧的处理逻辑。几个值得调的点参数建议说明单批大小100 ~ 500太小调度频繁太大内存和事务压力高sharding-total-count大于等于实例数保证每个实例至少一个分片streaming-process追积压用 true常规用 false流式会集中消化任务观察日志确认效果SQL 索引status create_time联合索引模运算无法走主键索引时至少让状态和时间条件走索引分片路由优先考虑主键区间或业务字段避免 MOD 取模在数据空洞下的倾斜另外我强烈建议在fetchData里加一行 debug 日志打印分片项和拉取条数。这个日志平时不显眼但排查流式循环问题时价值极高能直接告诉你“这个分片是否在工作”“当前是否在连续拉取”。提示MOD(id, n)这类函数条件会让 MySQL 放弃使用主键索引。数据量大时先把status和create_time的联合索引建好再考虑用 id 区间分片替代模运算。最后的个人实操体会Dataflow 作业我前后用过两年多最大的体会是它真正考验的不是你会不会写接口而是你有没有先想清楚数据的生命周期。一条数据从“待处理”变成“已处理”到底是哪个字段、哪条 SQL 完成的一旦想清楚流式模式、幂等、分片这些难题都能迎刃而解。另一个经验是上线前老老实实按三步走先用单实例、单分片跑通业务逻辑再开多实例验证分片不重复最后才开 streaming 模式追数据积压。跳过任何一步表面上省了时间实际上把风险堆到了线上。如果这篇文章对你有一点启发建议你现在就打开项目找一个可以数据驱动的任务试着用 Dataflow 重写一遍。不用追求一步到位先跑通再调优再优化分片策略这个过程中遇到的问题才是最有价值的学习材料。