企业官网建设流程全解析

1. 为什么需要Dataflow作业：从一次订单同步改造说起

做调度任务的人都知道，Elastic-Job里最常用的就是Simple作业，一个方法里写完所有业务逻辑，触发一次跑一次。但我运营过的一个订单系统，碰到了一个典型的"数据流"场景：每天有几万条待同步的订单，Cron每30秒触发一次，但每次触发时只需要处理当前积压的一部分，处理完一批继续拉下一批，直到数据全部清空。用Simple作业写当然能跑，但代码会非常别扭——你得把"拉数据"和"处理数据"两件事强行捏在一个方法里，批量大小写死，重试逻辑混乱，数据量一大就说不清楚到底是哪一步出了问题。

Dataflow作业就是为解决这类场景专门设计的。它把一次任务拆成两个明确阶段：先fetchData拉取一批待处理数据，再processData处理这批数据。两个方法各司其职，调度框架帮你在中间做协调，你不需要自己写循环，不需要手动控制批次边界，更不用在业务代码里维护"上一批处理到哪了"这种状态。最关键的，是它支持流式处理模式——数据不断产生时，框架会在处理完一批后自动继续拉取下一批，直到数据源为空才停下来。

为了把这个区别讲清楚，我用一个实际对比表来说明：

所以如果你手头的任务是"定时把一张表里的待处理数据搬走"、"消费消息队列里的积压消息"、"批量解析文件后写入目标库"，我建议直接考虑Dataflow作业，而不是在Simple作业里堆代码。它的模型和这类场景的思维方式是一致的，后期维护成本会低得多。这也是整个系列文章里我最推荐的作业类型之一，尤其是配合流式处理机制使用的时候。

2. 核心接口的语义拆解：fetchData与processData的"握手"约定

先来看Dataflow作业的标准接口定义，这是理解后续所有实现的基石：

public interface DataflowJob<T> extends ElasticJob { List<T> fetchData(ShardingContext shardingContext); void processData(ShardingContext shardingContext, List<T> data); }

泛型T就是你要处理的数据类型，可以是订单实体、日志对象、消息体，任何自定义类都可以。ShardingContext里带的是当前作业实例的分片上下文信息，shardingItem表示当前运行的是第几个分片，shardingTotalCount是总分片数，还有jobParameter和shardingParameter可以传递自定义参数。关于分片的配合，我后面专门展开讲，这里先聚焦两个方法本身的语义。

2.1 fetchData阶段：只负责问数据源要数据

fetchData的职责非常单一——返回一批待处理的数据。框架拿到这个方法返回的List<T>之后，会把整个列表原封不动传给processData。所以这个方法的实现要点就这么几个：

1. 返回的数据要控制在一个合理的批次大小内，比如100到500条，不要一次把几万条全捞出来；
2. 必须根据分片信息过滤数据，确保每个分片只处理自己负责的那部分数据；
3. 查询条件要能支撑重复调用，也就是多次调用fetchData时，已经处理过的数据不应该再出现。

我见过很多初学Dataflow的人把业务逻辑也写在fetchData里面，比如一边查数据一边更新状态。这是一个比较危险的习惯，数据获取阶段不应该有任何副作用，它只负责"读"。

2.2 processData阶段：交接与消化的地方

processData拿到的data参数，就是fetchData返回的那批数据。在这个阶段你可以做任何处理：写库、调用外部接口、发消息，什么都行。这一阶段的核心要求是"处理过程可重复执行且结果一致"，也就是我们常说的幂等性。因为调度框架在某些异常情况下会重新触发作业，如果processData不是幂等的，就会产生重复数据。

这里额外说明一个容易误解的细节：processData不像fetchData那样返回任何值，也就是说框架不会通过返回值来判断"这批数据处理成功没有"。如果你希望处理失败时数据不要丢，你需要在自己的代码里实现重试或补偿机制，或者在processData内部把异常捕获并记录下来，让作业继续往下跑。

2.3 两种执行模式的语义差异

这是Dataflow作业最核心的机制差异，也是很多人踩坑的地方。Elastic-Job为Dataflow作业提供了两种执行方式，由streamingProcess配置项控制：

• 非流式模式（streamingProcess=false）：每次Cron触发时，执行一次fetchData，把拿到的数据交给processData处理，然后本次触发结束。即使数据源里还有大量数据没处理完，也要等下一次Cron触发才会继续处理。
• 流式模式（streamingProcess=true）：Cron触发后执行fetchData，如果返回结果不为空，就调用processData；处理完成后框架会自动再次调用fetchData，再一次拿到数据就再处理，直到fetchData返回空列表（或null），本次触发才算结束。

用一句大白话概括：非流式是"触发一次处理一批"，流式是"触发一次能清空整个积压队列"。我会在第4章专门分析流式模式下的循环机制和风险控制，这里先记住这个语义差别就够了。

3. 一个最小可运行实现：Spring环境下的订单同步作业完整落地

理论讲再多，不如跑起来一个例子实在。下面我就以"订单状态同步"这个最常见的场景为例，从依赖引入、业务代码编写、配置项设置到验证运行，完整走一遍。这个案例是模拟项目，我尽量贴合真实生产环境。

3.1 引入Elastic-Job依赖

如果你用的是Spring Boot项目，建议直接引入elastic-job-lite-spring，它同时包含了核心库和Spring相关支持：

<dependency> <groupId>com.dangdang</groupId> <artifactId>elastic-job-lite-core</artifactId> <version>2.1.5</version> </dependency> <dependency> <groupId>com.dangdang</groupId> <artifactId>elastic-job-lite-spring</artifactId> <version>2.1.5</version> </dependency>

依赖里对应的高版本可能跟随不同分支发布，建议以实际可用的稳定版本为准。另外，Elastic-Job运行需要ZooKeeper作为注册中心，本地开发环境至少得有一个单机ZooKeeper实例，默认端口2181。

3.2 编写Dataflow作业类

我定义一个订单同步的作业类，核心逻辑是：从订单表中拉取待同步订单，调用外部同步服务的接口写入目标系统。

@Component public class OrderSyncDataflowJob implements DataflowJob<Order> { private static final int BATCH_SIZE = 200; @Autowired private OrderMapper orderMapper; @Autowired private OrderSyncService syncService; @Override public List<Order> fetchData(ShardingContext shardingContext) { int shardItem = shardingContext.getShardingItem(); int shardTotal = shardingContext.getShardingTotalCount(); // 只拉取属于当前分片的待同步订单 return orderMapper.selectPendingOrders(shardItem, shardTotal, BATCH_SIZE); } @Override public void processData(ShardingContext shardingContext, List<Order> data) { for (Order order : data) { boolean success = syncService.syncOrder(order); if (success) { orderMapper.markSynced(order.getId()); } else { // 记录失败日志，等待周期性重试 logger.warn("订单{}同步失败，稍后重试", order.getId()); } } } }

注意fetchData里的查询SQL，它是最需要纪律的部分。我通常用MOD取模的方式把数据均匀分到各个分片：

SELECT id, user_id, order_status, create_time FROM t_order_sync WHERE sync_status = 0 AND MOD(id, #{shardTotal}) = #{shardItem} ORDER BY id LIMIT #{batchSize}

这里的分片过滤逻辑非常关键。如果不加MOD(id, shardTotal) = shardItem这个条件，在多个实例同时运行时，每个实例都会捞到同一批数据，导致同一订单被处理多遍。我见过不少团队线上出问题时，查下来都是漏了这一步，所以印象特别深刻。

3.3 在Spring配置中注册作业

注册方式有XML和Java两种。XML方式在传统Spring项目里很常见：

<?xml version="1.0" encoding="UTF-8"?> <beans xmlns="http://www.springframework.org/schema/beans" xmlns:job="http://www.dangdang.com/schema/elastic-job-lite" xsi:schemaLocation="http://www.dangdang.com/schema/elastic-job-lite http://www.dangdang.com/schema/elastic-job-lite/elastic-job-lite.xsd"> <!-- 配置注册中心 --> <job:reg-center id="regCenter" server-lists="localhost:2181" namespace="elastic-job-demo" /> <!-- 注册Dataflow作业 --> <job:dataflow id="orderSyncDataflowJob" class="com.example.job.OrderSyncDataflowJob" cron="0/30 * * * * ?" sharding-total-count="2" streaming-process="true" /> </beans>

Java配置方式在Spring Boot项目里更常用，需要手动初始化调度器：

@Configuration public class ElasticJobConfig { @Bean public CoordinatorRegistryCenter registryCenter() { ZookeeperConfiguration zkConfig = new ZookeeperConfiguration("localhost:2181", "elastic-job-demo"); CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(zkConfig); regCenter.init(); return regCenter; } @Bean public JobScheduler orderSyncJobScheduler() { DataflowJobConfiguration jobConfig = new DataflowJobConfiguration( OrderSyncDataflowJob.class.getName(), 2, // shardingTotalCount "0/30 * * * * ?", // cron true // streamingProcess ); LiteJobConfiguration liteJobConfig = LiteJobConfiguration.newBuilder(jobConfig) .overwrite(true) .build(); JobScheduler scheduler = new JobScheduler(registryCenter(), liteJobConfig); scheduler.init(); return scheduler; } }

这里我建议把overwrite设置成true。开发阶段改作业参数很频繁，如果不覆盖，老的配置会一直在ZooKeeper里生效，你这边改半天没反应，排查起来会很沮丧。

3.4 启动与验证

作业启动后，可以在日志中看到类似Job 'orderSyncDataflowJob' initialized successfully的提示。等第一个Cron触发时间到达，日志里会出现fetchData执行记录和processData执行记录。我建议在fetchData里临时加一行日志打印本次拉取条数和当前分片编号，确认分片数据分配是否正确，确认无误后再移除。

一个相对省事的验证思路是：往订单表里插入2000条待同步数据，观察日志中处理记录，确认数据是在两次分片之间均衡分配、并且处理完成后表里的sync_status都变成了1。这个步骤能帮你把分片和流式处理两个核心逻辑一次性验证通过。

4. 流式处理是双刃剑：无限循环的成因与出口设计

流式处理这么强大，代价是什么？答案是它特别容易把自己跑成一个停不下来的死循环。这个章节我想用大量篇幅讲清楚它，因为它是我见过崩溃率最高的配置，没有之一。

4.1 无限循环是怎么发生的

在流式模式下，Elastic-Job的执行循环是这样的：

Cron触发 → 调用fetchData() → 如果返回数据非空：调用processData(数据) → 再次调用fetchData() → 如果返回数据非空：再次调用processData(数据) → ...持续循环... → 直到某次fetchData返回空列表（或null），本次执行结束

问题就出在"持续循环"这个机制上。如果fetchData每次都能查到数据，作业就永远不会停下来。一个错误的实现会很快暴露这个问题，比如：

@Override public List<Order> fetchData(ShardingContext context) { // 危险！每次查询都返回同样的数据 return orderMapper.selectAllPendingOrders(); }

如果processData没有把订单状态改成"已处理"，fetchData下一次又会捞到同一批数据，于是处理一批、捞一批、处理一批、捞一批，CPU和数据库被无限拖住。更隐蔽的情况是：fetchData每次确实能捞到新的数据，比如上游一直在写入，导致这个作业像沾上了强力胶一样，永远"处理不完"，一直占用一个线程，还能把任务堆积成雪崩。

4.2 什么样的"出口条件"是合格的

要避免无限循环，本质上要做到一点：每一次fetchData到processData的循环，都必须让待处理数据集朝"减少"的方向变化。具体到代码层面，我整理了几个切实可行的策略：

• 即时状态更新：在处理完成后立刻更新数据状态（比如把sync_status从0改成1），让fetchData下次查不到已处理数据。这是最基础的退出机制，但要注意状态的更新必须和业务处理在同一个事务里，或者至少保证失败时不更新状态，否则会出现"处理一半，状态却显示已完成"的数据空洞。
• 按批次时间窗口处理：fetchData每次只取"最近5分钟内进入"的数据，处理完这批，下一次继续取下一个5分钟窗口。这种方式天然有界，但要注意时间窗口和前一次窗口的边界重叠问题。
• 限量退出保护：在fetchData里维护一个本批次循环次数的计数，达到上限就返回null。这种方式适合数据量特别大、无法在单次触发中处理完的场景。不过要小心计数器的生命周期，不要跑到下次触发还在累加，否则会导致某些批次被跳过去。

我自己总结了一条经验法则：如果fetchData里没有任何"拉取条件随处理进度变化"的逻辑，那大概率会出问题。一个合格的fetchData，它在不同循环轮次中查询到的数据集合应该逐渐减小，直到变成空集合。

4.3 批次大小和线程占用的经验值

流式处理模式下，批次大小直接决定每个循环的速度和资源占用。我通常建议单批控制在200到500条，不宜超过1000条。批次太大，processData单轮处理时间变长，框架在这段时间内无法响应新的Cron触发，而且一旦中间出错，重放的成本会很高；批次太小，循环调度太频繁，日志刷得厉害，数据库查询开销也大。

还有一个容易被忽视的点：流式作业在长时间运行时，线程池资源一直被占用。如果你的作业分片数是4，理论上最多同时有4个分片线程在跑。这时候如果还有别的Simple作业也要用同一个调度线程池，要注意线程池大小是否足够，否则其他作业会排队等待，表现为"触发时间到了不执行"。

4.4 流式模式一定比非流式更好吗

不一定。如果数据积压量本身不大，或者业务上能接受"一次Cron处理一批、剩下的下个周期继续"这种节奏，非流式会更省心，因为它的每次触发都干净利落，不存在循环风险。流式模式的收益只在"数据持续产生并且要求尽快消费完"的场景下才值得承担额外的风险。选型时不要盲目开流式，先想清楚数据量的实际情况。

5. 分片协同不是可选项：数据过滤策略与幂等兜底

Dataflow作业天然适合配合分片使用，但分片逻辑写错了，数据处理就会出大乱子。这个章节专门讲清楚分片和数据分配之间的配合细节。

5.1 分片上下文里到底有什么

每个作业实例启动后，Elastic-Job会根据分片策略把总分片数分配给各个实例。比如总分片数是2，两个实例各拿一个分片；如果只有一个实例，它会同时拿两个分片。在fetchData和processData方法里拿到的ShardingContext，就包含了当前这个分片的具体信息。

ShardingContext { jobName: "orderSyncDataflowJob", shardingTotalCount: 2, shardingItem: 0, // 当前实例处理第0号分片 shardingParameter: "", // 分片自定义参数 jobParameter: "" // 作业自定义参数 }

shardingItem永远从0开始编号，最大值是shardingTotalCount - 1。这个shardingItem就是你在SQL里用来过滤数据的依据。

5.2 两种常见的数据过滤策略

第一种是取模分片，也就是我在第3章示例里写的MOD(id, shardTotal) = shardItem。它实现简单，数据均匀分布，缺点是如果主键不是连续自增或者删改频繁，分布可能略有不均。这个策略适合数据量大、对均衡性要求不高的场景。

第二种是基于时间范围的分片，适合按业务日期划分数据的场景。比如你可以配置shardingParameter为不同日期，然后在fetchData里根据shardingParameter拼装查询条件。这种方式的优点是语义直观、数据边界清晰，缺点是需要自己规划好每个分片的数据范围。

我见过一种比较隐蔽的Bug：有人直接把shardingTotalCount当成"并行线程数"，在配置中心里随意调整。每次调整后，由于取模的基数变了，原本分配给分片0的数据可能被分到分片1，如果处理状态没有及时落库，调整分片数量的那一刻会发生小范围的重复处理。所以分片数量前期规划好以后，不要频繁改动，改之前要确认数据兼容性。

5.3 幂等兜底是最后一道防线

即使分片逻辑写对了，也不能保证数据百分之百不会被重复处理。比如作业因为网络抖动被重新触发、故障转移时某个分片被从实例A迁移到实例B、处理过程中进程崩溃后重启，这些情况都可能导致同一条数据被processData处理两遍。

所以我的实践原则是：分片保证"正常情况下不重不漏"，幂等设计兜底"异常情况下不产生脏数据"。具体做法很简单，在处理数据时用主键或唯一业务键做去重判断，比如在processData里先查一遍目标系统是否已存在这条订单，或者在目标表加唯一索引，重复插入直接捕获异常跳过。这些兜底逻辑虽然每个看起来多花了几行代码，但在生产环境上能帮你避免很多半夜被叫起来的报警。

6. 我在生产环境踩过的三个Dataflow作业的坑

最后分享三个我真实遇到过的问题。每个问题我都尽量按"现象—排查—根因—修复"的顺序写清楚，希望能帮你看完少走弯路。

6.1 流式作业"永远处理不完"：数据库CPU飙升

现象：一个订单同步作业某个晚上开始，数据库CPU持续接近100%，作业日志刷个不停，每次fetchData都能查到几千条数据，processData也一直在处理，但表里待同步的数据量却一直没下降。

排查过程：先看fetchData的查询SQL，发现只查了sync_status = 0，而processData里虽然调用了同步接口，但只有同步成功才把状态改成1。问题出在同步接口的返回结果大部分是"失败"状态，失败的数据没有更新状态，于是下一秒fetchData又会把这些失败记录捞出来，反复循环重试。这本质上不是无限循环Bug，而是失败重试没有退避策略导致的"无效循环"。

修复方案：在processData里增加失败重试的退避逻辑——记录失败次数和下次重试时间，fetchData只查"到了重试时间"的数据，同时在单次循环里限制最大处理条数，从源头避免无效循环。

6.2 分片数从2改成4之后，数据重复了一部分

现象：某次数据迁移需要临时增加实例，我把分片数从2调成了4。调整后当晚，业务侧反映出现大量重复订单同步记录。

排查过程：查日志发现重复的订单主要集中在原分片0和数据区间的边界。根因是取模分片在分片数从2变4时，数据归属被重新划分，而历史数据处理状态已经更新完毕。原本属于分片0的某些数据，调整后被划分到分片1或分片2，新分片从0开始跑，发现这些数据还是"待处理"状态，于是又重新处理了一遍。

修复方案：临时在同步逻辑里增加幂等判断，重复执行时直接跳过，同时把分片数改动安排在业务低峰期，并且改动后进行一次全量对账。分片数这个参数，真不是随便能动的。

6.3 关闭事件追踪时忘了关streamingProcess的日志刷屏

现象：一个内部报表任务，数据量不算大，但每次运行都会产生巨量Elastic-Job内部日志，日志文件一小时能涨500MB。

排查过程：这个作业配置了事件追踪和流式处理。流式模式下每轮循环都会产生多次fetch和process的事件记录，事件追踪把每个事件都写入库表或日志文件。数据量小的时候看不出来，数据量一大，事件数据反而比业务数据还多。

修复方案：对这批高频流式作业关闭事件追踪，或者只保留关键异常日志。事件追踪功能很实用，但不是所有作业都适合开，尤其流式模式下事件量会成倍放大，要有取舍。这也提醒我，任何"看似无害"的监控开关，放在高频率循环里，都会被放大成资源黑洞。

写到这里，Dataflow作业的实现要点基本都覆盖了。最后再分享一个我自己的习惯：凡是核心生产作业，都先在本地构造一个能控制数据量的模拟环境，把流式模式打开跑一遍，观察fetchData的轮询频率和退出时机是否符合预期。确认出口条件稳定、分片分配正确、重复处理兜底生效之后，再部署到生产。这套验证流程帮我拦下了不少潜在事故，你也不妨试试。