企业官网建设流程全解析

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) { // 查超时订单 List<OrderEntity> 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 里写 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 DataflowJob<OrderEntity> { @Override public List<OrderEntity> fetchData(ShardingContext shardingContext) { // 按分片条件捞一批待处理数据 return null; } @Override public boolean processData(ShardingContext shardingContext, List<OrderEntity> 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 DataflowJob<OrderEntity> { @Override public List<OrderEntity> fetchData(ShardingContext shardingContext) { // 按分片条件捞一批待处理数据 return null; } @Override public void processData(ShardingContext shardingContext, List<OrderEntity> 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-process=false（默认值）时，每次 cron 触发后，框架只执行一次fetchData+processData，不管取到多少数据，处理完这一批就结束，等下一个调度周期。适合数据量不大、单批处理时间可控的常规场景。

streaming-process=true时，框架进入流式循环：取一批 → 处理一批 → 再取一批 → 再处理，直到fetchData返回空列表才结束本次触发。适合需要“追积压”的场景，比如每分钟补偿一次，每次都尽可能把当前积压的清空。

一个常见误区是认为 streaming 模式性能更好。其实它只是改变循环控制方式，并不改变你单批处理的速度。它的价值在于：一次触发内能连续消化多批数据，而不是每批都要等下一个 cron 周期，这对积压追赶非常有效。

我在生产上踩过一次坑：明明只是普通的定时状态扫描，数据量也不大，却把streaming-process设成了true，线上任务偶尔会出现“一次触发跑了几分钟”的现象，表面看是性能问题，其实是流式循环把本应分摊到多个周期的活集中在一次做完了。所以我的建议是：默认先开 false，确有必要追数据积压时再开 true。

2.4 ShardingContext 怎么参与数据路由

ShardingContext 是 Dataflow 作业里绕不开的对象，它提供了几个关键信息：

最常规的用法是配合数据库取模来路由。比如总分片数为 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 { List<OrderEntity> 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 id="selectTimeoutOrders" resultType="OrderEntity"> 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} </select>

3.2 编写 DataflowJob 实现类

作业实现类代码如下，2.x 版本：

public class OrderCloseDataflowJob implements DataflowJob<OrderEntity> { private static final int BATCH_SIZE = 200; private final OrderMapper orderMapper; public OrderCloseDataflowJob(OrderMapper orderMapper) { this.orderMapper = orderMapper; } @Override public List<OrderEntity> fetchData(ShardingContext shardingContext) { return orderMapper.selectTimeoutOrders( shardingContext.getShardingItem(), shardingContext.getShardingTotalCount(), BATCH_SIZE); } @Override public boolean processData(ShardingContext shardingContext, List<OrderEntity> data) { for (OrderEntity order : data) { orderMapper.markClosed(order.getId(), LocalDateTime.now()); } return true; } }

3.x 版本只需要把processData的返回值改成void：

@Override public void processData(ShardingContext shardingContext, List<OrderEntity> 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 方式。这是新项目最推荐的写法，配置清晰、不依赖 Spring：

ZookeeperConfiguration 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 id="orderCloseJob" class="com.example.job.OrderCloseDataflowJob" registry-center-ref="regCenter" cron="0 */1 * * * ?" sharding-total-count="4" streaming-process="true" />

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应该变成 2，close_time有值。

第三步观察流式循环。配置了streaming-process=true时，日志里会连续出现多轮fetch size=200、fetch size=100，最后出现一次fetch empty，表示本次触发结束。如果只看到一轮 fetch 就结束了，要么是数据不够，要么是配置没生效，可以去注册中心确认作业属性。

第四步验证多实例不重复。两个实例同时运行时，同一订单只会被其中一个实例关闭。判断标准：MOD(id, 4)的结果只属于一个实例，处理日志中不会出现两条相同的订单 id。

4. 常见问题与排查技巧实录

4.1 streaming 模式下被“死循环”卡住

这是 Dataflow 作业最经典的事故。表现是：作业一直在跑，日志里不断打印 fetch 到数据、处理数据，数据库里某些记录反复出现，作业线程被占住不放。

我当时的排查思路是：死循环的本质一定是“处理完的数据，仍然满足 fetchData 的查询条件”。具体原因通常有几种：

• processData只做了逻辑判断，没有更新查询条件里的状态字段；
• 更新的是另一张关联表，而 fetchData 查的是主表；
• 更新语句的 WHERE 条件和查询语句不一致，导致某些记录永远改不动；
• 处理抛了异常被吞掉，状态没更新，但循环还在继续。

排查步骤我整理成速查表：

一个实用的兜底方案：在fetchData的 SQL 里加一个处理次数限制，比如额外加一个retry_count < 5的条件，避免单条脏数据把整个作业拖死。

4.2 分片不均导致的数据漏处理

用MOD(id, shardingTotalCount)做路由，最常见的问题是 id 不连续导致分片倾斜。比如订单表里大量订单是通过中间表导入的，id 分布不均，结果某个分片的数据特别多，另一个分片几乎没有数据，表现为“某些实例忙死，某些实例闲死”。

如果是漏处理，问题更隐蔽：你以为 4 个分片覆盖了全表，但MOD的分布和实际数据分布不一致，某个值得处理的数据恰好落在没有数据的分片里。排查的方法是分别把shardingItem=0到shardingItem=3的查询结果数量统计出来，看是否符合预期。

对于数据量大的表，我更推荐用主键区间切分，而不是模运算。思路是先用一条 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.DataflowJob，3.x 改成了org.apache.shardingsphere.elasticjob.api.dataflow.DataflowJob，IDE 直接全红，只能批量替换。

第二个是processData返回值变化。刚才提到过，2.x 返回 boolean，3.x 返回 void。如果你的老代码里维护了一个return false的分支，迁移时千万不要直接删掉或改成return，要先想清楚这个分支原来想表达什么，改成抛出业务异常或者记录日志。

第三个是作业启动方式变化。2.x 用JobScheduler+LiteJobConfiguration，3.x 用ScheduleJobBootstrap（定时作业）和OneOffJobBootstrap（一次性作业）。对于 Dataflow 这种持续型作业，用ScheduleJobBootstrap就行。

第四个是 Spring 配置方式变化。2.x 的 XML namespace 在 3.x 不再推荐，官方主推注解和 Java API。如果你项目里大量用了 XML 配置，迁移前先评估改动成本。

4.5 性能调优的几个参数

Dataflow 作业的性能瓶颈不在框架，而在你 fetchData 的 SQL 和生产侧的处理逻辑。几个值得调的点：

另外，我强烈建议在fetchData里加一行 debug 日志，打印分片项和拉取条数。这个日志平时不显眼，但排查流式循环问题时价值极高，能直接告诉你“这个分片是否在工作”“当前是否在连续拉取”。

提示：MOD(id, n)这类函数条件会让 MySQL 放弃使用主键索引。数据量大时，先把status和create_time的联合索引建好，再考虑用 id 区间分片替代模运算。

最后的个人实操体会

Dataflow 作业我前后用过两年多，最大的体会是：它真正考验的不是你会不会写接口，而是你有没有先想清楚数据的生命周期。一条数据从“待处理”变成“已处理”，到底是哪个字段、哪条 SQL 完成的，一旦想清楚，流式模式、幂等、分片这些难题都能迎刃而解。

另一个经验是上线前老老实实按三步走：先用单实例、单分片跑通业务逻辑；再开多实例验证分片不重复；最后才开 streaming 模式追数据积压。跳过任何一步，表面上省了时间，实际上把风险堆到了线上。

如果这篇文章对你有一点启发，建议你现在就打开项目，找一个可以数据驱动的任务，试着用 Dataflow 重写一遍。不用追求一步到位，先跑通，再调优，再优化分片策略，这个过程中遇到的问题，才是最有价值的学习材料。