为什么需要“DAG+Quartz”的分布式任务调度？

在我们日常开发过程中，任务调度始终是支撑核心业务流程的“隐形引擎”。无论是电商大促时的“库存扣减→订单生成→支付通知→物流同步”，还是数据处理平台的“日志采集→清洗→聚合→分析”，或是金融系统的“交易对账→风控校验→报表生成”，都需要任务按特定顺序、依赖关系高效执行。

传统定时任务工具（如Quartz）虽能解决“何时执行”的问题，但面对复杂业务场景时，逐渐暴露出三大痛点：

• 依赖关系难管理：任务间若存在“前置任务未完成则后续任务不能执行”的强依赖（如支付成功后才能发短信），传统工具无法直观描述这种逻辑，只能通过“延迟触发”或“人工干预”勉强实现，效率低且易出错；
• 并行执行难实现：对于无依赖关系的任务（如大促时同时执行“库存扣减”和“订单生成”），传统工具需手动拆分任务或依赖外部协调，难以动态适配业务变化；
• 失败处理难闭环：任务失败后，传统工具仅支持简单的“重试”或“标记失败”，无法根据失败类型（如偶发网络问题 vs 永久性错误）动态调整策略（如指数退避重试 vs 熔断跳过），导致资源浪费或业务中断。

而 DAG（有向无环图） 恰好能通过“节点表示任务、边表示依赖”的模型，清晰描述任务的执行逻辑。结合Quartz的分布式调度能力，可构建一个“能描述依赖、支持并行、灵活容错”的任务调度系统。但现实中，这一目标的实现面临多重挑战：

• DAG与Quartz的集成复杂度高：Quartz的任务触发逻辑需与DAG的拓扑排序结果深度绑定，如何动态生成任务执行顺序并触发？
• 分布式环境下的一致性难题：多实例部署时，如何避免任务重复触发？如何保证任务状态（如“执行中”“成功”“失败”）在分布式系统中的准确同步？
• 失败处理的工程化落地：如何设计可配置的重试策略（如指数退避）？如何通过熔断机制避免“失败任务反复重试”导致的资源耗尽？

本文将围绕这些核心问题，从“环境搭建→核心功能开发→测试验证”全程拆解，手把手实现一个支持“依赖管理、并行执行、动态容错”的分布式任务调度系统，解决传统工具无法应对的复杂场景需求。

一、环境准备

1.1 必须的软件和环境

1.1.1 JDK 1.8

• 下载地址：https://adoptium.net/temurin/releases/?version=8（选择Windows/Linux/macOS版本）。
• 验证安装：命令行输入 java -version，输出类似 java version "1.8.0_301" 表示成功。

1.1.2 Maven 3.8+（管理Java依赖）

• 下载地址：https://maven.apache.org/download.cgi（选择二进制包）。
• 安装步骤（以Linux为例）：

1.1.3 MySQL 8.0（存储Quartz的元数据）

• 下载地址：https://dev.mysql.com/downloads/mysql/（选择MySQL Community Server 8.0.x）。
• 安装步骤（以Linux为例）：

1.1.4 Redis 7.0+（存储任务状态和分布式锁）

• 下载地址：https://download.redis.io/releases/redis-7.0.12.tar.gz。
• 安装步骤（以Linux为例）：

1.2 创建Spring Boot项目

1.2.1 访问Spring Initializr

打开浏览器，访问 https://start.spring.io/，按以下步骤创建项目：

1.2.2 导入项目到IDE

• 使用IntelliJ IDEA或Eclipse导入Maven项目（选择 pom.xml 文件）。

1.3 配置数据库和Quartz集群（关键！）

1.3.1 创建MySQL数据库

登录MySQL，创建存储Quartz元数据的数据库：

1.3.2 配置Quartz集群（必须！）

在 src/main/resources/application.yml 中添加以下配置：

1.3.3 初始化Quartz表（自动完成）

首次启动Spring Boot应用时，Quartz会自动在 quartz_db 数据库中创建所需的表（如 QRTZ_TRIGGERS、QRTZ_CRON_TRIGGERS 等）。启动后，登录MySQL验证：

二、核心功能开发

2.1 定义DAG的数据结构（JSON格式）

DAG（有向无环图）用JSON格式描述任务关系，包含节点（任务）和边（依赖关系）。例如，一个“订单支付后通知”的DAG定义如下：

关键说明：

• dagId：DAG的唯一标识（如 order_pay_notify），用于区分不同的任务链路；
• nodes：任务列表，每个任务有 id（唯一ID）、name（显示名称）、jobClass（任务类的全限定名）；
• edges：依赖关系，from 是前置任务ID，to 是后续任务ID（表示“前置任务完成后才能执行后续任务”）。

2.2 解析DAG：将JSON转为Java对象

需要将JSON格式的DAG定义转换为Java对象，方便程序处理。

2.2.1 定义DAG的节点（TaskNode）和边（TaskEdge）

在 src/main/java/com/example/dag/entity 目录下创建以下类：

2.2.2 定义DAG整体结构（DagDefinition）

2.2.3 解析JSON的DAG解析器（DagParser）

使用Fastjson将JSON字符串转换为 DagDefinition 对象：

2.3 拓扑排序：生成任务的执行顺序

DAG的任务必须按依赖关系执行（比如“支付校验”完成后才能执行“发送短信”）。我们需要用 拓扑排序 算法（Kahn算法）生成执行顺序。

2.3.1 拓扑排序的原理

拓扑排序的核心是：

1. 统计每个任务的入度（有多少前置任务）；
2. 将入度为0的任务（没有前置任务）加入执行队列；
3. 依次执行队列中的任务，并将其后续任务的入度减1；
4. 重复步骤2-3，直到所有任务都被执行。

如果有任务未被执行，说明DAG中存在循环依赖（比如A依赖B，B依赖A），此时拓扑排序会失败。

2.3.2 实现拓扑排序（TopologySorter）

在 src/main/java/com/example/dag/service 目录下创建 TopologySorter.java：

2.4 集成Quartz：触发任务执行

Quartz是Java生态中最经典的任务调度框架，支持定时触发和分布式执行。我们需要让Quartz根据拓扑排序的结果，按顺序触发任务。

2.4.1 自定义Quartz任务类（BaseJob）

所有任务需要继承Quartz的 QuartzJobBean 类，并实现 executeInternal 方法（任务的具体逻辑在这里编写）。

在 src/main/java/com/example/dag/job 目录下创建 BaseJob.java：

2.4.2 示例任务类（发送短信）

编写一个具体的任务类（例如“发送短信”任务），继承 BaseJob 并实现 executeTask 方法：

2.4.3 触发DAG执行（DagScheduler）

创建一个调度器类，负责解析DAG、生成执行顺序，并调用Quartz触发每个任务：

2.5 状态管理：记录任务的执行状态

为了知道任务是否成功或失败，需要记录每个任务的执行状态。这里用Redis存储状态（方便分布式访问）。

2.5.1 定义任务状态枚举（TaskState）

在 src/main/java/com/example/dag/enum 目录下创建 TaskState.java：

2.5.2 状态管理器（StateManager）

使用Redis存储每个任务的状态，并提供更新状态的方法：

2.6 失败处理：重试与熔断

任务执行失败时，需要自动重试或触发熔断机制，避免无限失败。

2.6.1 失败重试策略（RetryPolicy）

定义一个简单的指数退避重试策略（失败后等待时间逐渐增加）：

2.6.2 熔断机制（CircuitBreaker）

当某个任务频繁失败（比如失败率超过50%），触发熔断，暂时跳过该任务：

三、测试验证

3.1 启动系统

1. 确保MySQL和Redis已启动；
2. 在Spring Boot项目的根目录执行 mvn spring-boot:run，启动应用；
3. 控制台输出 Started DagSchedulerDemoApplication in X seconds 表示启动成功。

3.2 触发DAG执行

通过HTTP接口触发DAG执行（需要先添加一个Controller）：

3.3 发送POST请求触发DAG

使用Postman或curl发送以下请求（替换 dagJson 为你的DAG定义）：

3.4 查看任务执行结果

3.4.1 查看控制台输出

任务执行时，控制台会打印每个任务的触发信息（如“触发任务：check_pay”）。

3.4.2 查看Redis状态

使用Redis客户端（如 redis-cli）查看任务状态：

四、常见问题与解决

4.1 Quartz集群启动失败

现象：启动应用时报错 org.quartz.JobPersistenceException: Couldn't store trigger...。
原因：Quartz的元数据表未正确创建。
解决：手动执行Quartz的建表脚本（https://github.com/quartz-scheduler/quartz/tree/main/quartz-core/src/main/resources/org/quartz/impl/jdbcjobstore），或检查MySQL用户是否有建表权限。

4.2 任务未按顺序执行

现象：“发送短信”任务在“支付校验”完成前就执行了。
原因：拓扑排序逻辑错误，或Quartz触发任务时未等待依赖。
解决：检查 TopologySorter 的排序逻辑，确保入度计算正确；或在触发任务时，检查前置任务是否已完成（通过 StateManager 查询状态）。

4.3 任务失败后未重试

现象：任务执行失败后，状态直接变为 FAIL，没有重试。
原因：未在任务中调用重试逻辑，或 RetryPolicy 未正确配置。
解决：在 BaseJob 的 executeInternal 方法中，捕获异常后调用重试逻辑（如使用 RetryPolicy 计算间隔，重新触发任务）。

总结

到此为止，已经完成了一个支持“任务依赖、并行执行、失败重试”的分布式任务调度系统的搭建。核心流程是：

1. 用JSON定义DAG的任务关系；
2. 用拓扑排序生成执行顺序；
3. 用Quartz触发任务执行；
4. 用Redis记录任务状态；
5. 用重试和熔断机制提升健壮性。

当然，这只是一个基础实现，实际生产环境中还可以扩展：

• 增加任务参数传递（上游任务的输出作为下游任务的输入）；
• 提供可视化界面（如前端拖拽编辑DAG）；
• 集成监控告警（任务失败时发送邮件/钉钉通知）；
• 支持动态修改DAG（运行时调整任务链路）。

最后，希望通过本文，你能大致了解DAG+Quartz整个流程，后续可以在此基础上进一步优化探索！