easy-schedule 组件设计

状态：方案已确认，v1 示例代码与集成测试已就绪（2026-09-04）范围：cron 周期 + 一次性/延迟任务；分片并行（v1.1）；DAG（v2，jgrapht 编排）引擎：db-scheduler 16.12.0（com.github.kagkarlsson，Apache-2.0，要求 Java 17+，与框架 JDK17 升级匹配）原则：纯嵌入式 SDK、不引入 ZooKeeper/Redis 作为任务存储、不新增独立服务进度：@Job 注解 / JobRegistry / JobInvoker / SchedulerRuntime / ScheduleManager 已实现；example + 集成测试（HSQLDB 内存库）6/6 通过。MySQL 生产 DDL 见 src/main/resources/schema/mysql-scheduled-tasks.sql（需配合 easy.schedule.always-persist-timestamp-in-utc=true）。

一、选型结论

决策点	结论	理由
引擎	db-scheduler	嵌入式 + 集群单实例执行（DB 行锁）+ cron/一次性开箱即用 + 仅 slf4j 依赖 + Apache-2.0；与框架”轻量、自管理、无外部中间件”哲学一致
协调载体	现有 MySQL	ZooKeeper 过重；Redis 不适合做任务持久化；Kafka KRaft 为 Kafka 内部协议不可复用
分片模型	框架层建模（cron 广播 → N 个一次性条目）	db-scheduler 本身无分片概念，但”一次性任务跨节点并发”天然满足分片并行
任务定义存储	声明式（注解，随代码版本化）	与 AVRO 建模 `@Scheduler(cron)` 一致；v1 免管理后台
一次性任务/执行实例	DB 表（引擎托管）	运行时动态产生，需持久化、可恢复、可查
DAG	v2 用 jgrapht 做拓扑，独立于引擎	编排是框架职责，不绑架调度引擎

候选排除记录

XXL-Job / PowerJob / DolphinScheduler：中心化平台形态，需独立部署调度中心，与”嵌入式 SDK”诉求冲突。
ElasticJob-Lite：分片/故障转移最强，但集群强制依赖 ZooKeeper 3.6+，运维面在组件之外。
JobRunr：功能全但 DAG/批处理仅 Pro 商业版，社区 LGPL，许可证与功能分裂不合适。
全自研：无现有引擎承担”单实例执行/持久化/补偿”，重复造轮子。

二、总体架构

flowchart TD
    subgraph "easy-schedule (embedded)"
        A["@Job 注解处理器\nSchedulerAnnotationProcessor"] --> B["ScheduleManager 门面"]
        C["一次性/延迟 API\nScheduleManager.fireOnce()"] --> B
        B --> D["DbScheduler 引擎\n(db-scheduler, MySQL 单表)"]
        D --> E["JobExecutor"]
        E --> F["JobRegistry\njobCode -> Spring Bean 方法"]
        E --> G["ExecutionContext\n(shardIndex/shardTotal/payload)"]
        D --> H["FailureHandler 失败策略"]
        B --> I["ShardBroadcaster 分片广播 (v1.1)"]
        I --> D
        J["DagDriver DAG 编排 (v2, jgrapht)"] --> D
    end
    D --> K["scheduled_tasks 表 (MySQL)"]
    E --> L["业务 @Job 方法"]

三、核心概念与 API 草图

3.1 统一任务注解 `@Job`

一个注解同时支撑”周期 / 一次性 / 分片”三种触发方式，方法即处理器，避免多套注解体系：

@Target(ElementType.METHOD)
@Retention(RetentionPolicy.RUNTIME)
public @interface Job {
    String name();              // jobCode，全局唯一
    String cron() default "";   // 空 = 仅支持 API 触发/一次性
    String desc() default "";
    int shardCount() default 1; // >1 时 cron 触发后广播 shardCount 个分片任务
}

处理器方法签名约定（按需注入，与 Spring @Scheduled 风格一致）：

@Component
public class ReportJob {
    @Job(name = "dailyReport", cron = "0 0 2 * * *", desc = "每日报表", shardCount = 4)
    public void handle(ExecutionContext ctx) {
        // ctx.shardIndex() / ctx.shardTotal() 决定本分片处理的数据段
        // ctx.payload() 一次性任务携带的业务参数
    }
}

3.2 门面 API

public interface ScheduleManager {
    // 一次性/延迟：按 jobCode 触发一次（可携带 payload）
    String fireOnce(String jobCode, Object payload);
    String fireOnce(String jobCode, Object payload, LocalDateTime runAt);

    // 手动触发周期任务（等价跑一次广播/一次分片）
    String trigger(String jobCode);

    // 运行态注册/注销（非注解的编程式任务）
    void register(JobDescriptor descriptor);
    void unregister(String jobCode);
}

3.3 三种触发路径

路径	机制
cron 周期	`@Job(cron=...)` 在启动时注册为 db-scheduler `RecurringTask`，按 cron 到期，由引擎保证仅一个实例触发
一次性/延迟	`ScheduleManager.fireOnce` → 写入 db-scheduler 一次性任务条目，到期后任意空闲节点执行，失败可配重试
分片并行（v1.1）	cron 到期触发的执行体为 ShardBroadcaster：拆出 shardCount 个一次性条目（各自携带 shardIndex），并行被各节点消费；跨节点并行由”不同条目被不同 worker 抢占”天然达成

3.4 ExecutionContext

public interface ExecutionContext {
    String jobCode();
    int shardIndex();   // 默认 0
    int shardTotal();   // 默认 1
    <T> T payload();    // 一次性任务参数
    String instanceId();// 本次执行实例号（可追溯）
}

四、分片并行模型（v1.1）

语义：数据段分片（每分片处理一段数据），不做”分片固定钉死节点”的一致性保证——与 ElasticJob 的差异是设计取舍：多数”按 shardIndex 处理数据区间”的场景不需要固定节点；需要固定节点/故障精确接管时再评估升级路径。
流程：

flowchart TD
    A["cron 到期 (单实例触发)"] --> B["ShardBroadcaster 生成 N 个一次性条目"]
    B --> C["节点A 抢到 shard0"]
    B --> D["节点B 抢到 shard1"]
    B --> E["节点C 抢到 shard2"]
    C --> F["各分片业务处理 + 执行记录"]
    D --> F
    E --> F

每个分片条目独立失败/重试，互不影响；总分片数在注解中静态声明（v1），运行态动态调片数列入后续版本。

五、DAG 设计（v2，jgrapht）

定义：DagBuilder 用 jgrapht 的 DirectedAcyclicGraph 描述任务依赖（节点=jobCode，边=先后关系），构建时做环检测。
执行：DAG 实例化后持久化到框架自管表，驱动逻辑以”一次性任务”方式入队——每推进一批满足前置条件的节点（入度归零），节点完成回调驱动继续推进，直至全部终态。

DagBuilder.create("refreshWarehouse")
    .node("syncOrder").node("syncStock").node("computeMetrics")
    .dependsOn("computeMetrics", "syncOrder").dependsOn("computeMetrics", "syncStock")
    .build();  // 内部 jgrapht 环检测

DAG 与引擎解耦：节点仍是普通一次性任务，DAG 状态机（就绪/成功/失败/跳过）由框架自管表维护，天然支持分布式推进。

六、表结构

6.1 引擎表（db-scheduler 托管，DDL 随组件提供）

db-scheduler 需要一张 scheduled_tasks 表（各数据库有官方建表 SQL，MySQL 版随组件 resources 提供），字段含 task_name / task_handler / task_data / execution_time / picked / picked_by / last_heartbeat / version 等，由引擎全权管理，业务不直改。

6.2 框架自管表（可选，v1 提供）

-- 执行记录（审计/补偿查询，v1）
CREATE TABLE IF NOT EXISTS easy_schedule_exec_log (
    id           BIGINT AUTO_INCREMENT PRIMARY KEY,
    instance_id  VARCHAR(64)  NOT NULL,
    job_code     VARCHAR(128) NOT NULL,
    shard_index  INT          NOT NULL DEFAULT 0,
    status       VARCHAR(16)  NOT NULL,           -- RUNNING/SUCCESS/FAILED
    error_msg    VARCHAR(1024),
    start_time   DATETIME,
    finish_time  DATETIME,
    node_id      VARCHAR(64),
    KEY idx_job (job_code, status)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;

-- DAG 实例（v2）
CREATE TABLE IF NOT EXISTS easy_schedule_dag_instance (
    dag_code    VARCHAR(128) NOT NULL,
    instance_id VARCHAR(64)  NOT NULL,
    node_code   VARCHAR(128) NOT NULL,
    status      VARCHAR(16)  NOT NULL,            -- READY/RUNNING/SUCCESS/FAILED/SKIPPED
    PRIMARY KEY (instance_id, node_code)
) ENGINE = InnoDB DEFAULT CHARSET = utf8mb4;

七、模块与包结构

components/easy-schedule/
├── pom.xml                    # 引入 db-scheduler-spring-boot-starter(适配 SB 2.7)
└── src/main/java/io/matrix/schedule/
    ├── annotation/            # @Job
    ├── autoconfigure/         # ScheduleAutoConfiguration + ScheduleProperties(easy.schedule.*)
    ├── core/                  # ScheduleManager / ScheduleManagerImpl / JobRegistry / JobDescriptor
    ├── context/               # ExecutionContext / DefaultExecutionContext
    ├── executor/              # JobExecutor / JobInvoker(反射调 Spring Bean 方法) / FailureHandler
    ├── sharding/              # ShardBroadcaster (v1.1)
    └── dag/                   # DagBuilder / DagDriver / DagStateStore (v2)
└── src/main/resources/
    ├── META-INF/spring.factories          # 注册自动配置
    └── schema/mysql-scheduled-tasks.sql   # 引擎建表 DDL

自动配置遵循组件规范：@AutoConfigureAfter(DataSourceAutoConfiguration.class)，关键 Bean 均 @ConditionalOnMissingBean 允许业务覆盖。

八、与现有组件的关系

kernel-common 的 RedisScheduler：保留，定位”近实时延迟消息/内存级轻量延迟”，与 easy-schedule 的”持久化 cron/一次性”互补，不合并。
easy-lock：v1 不需要（单实例执行由引擎 DB 行锁兜底）；分片补偿/选主类需求若出现再复用。
easy-validate/easy-event-bus：不强依赖；失败通知可后续经 easy-event-bus 事件发布，预留扩展点即可。

九、边界与风险声明

项	说明
精度	基于 DB 轮询（默认秒级），不做毫秒级精确调度；需要近实时请走 RedisScheduler
一致性	保证”同一任务条目单实例执行”，不承诺脑裂级强一致（与 db-scheduler 语义一致）
分片节点分配	不固定分片-节点绑定（v1.1 语义），如需强绑定需引入协调服务，列为升级路径
建表	引擎表需人工执行一次 DDL（组件提供 SQL，不自动执行，避免越权改库）
版本	Spring Boot 2.7 采用 `db-scheduler-spring-boot-starter`，具体版本以引入时 Maven Central 匹配 SB 2.7 的最新稳定版为准（基线 12.x+，当前 16.x 活跃维护）

十、落地里程碑

v1.0：引擎接入 + @Job cron 注册 + fireOnce 一次性 API + JobRegistry + 失败策略 + 执行记录（可选）
v1.1：ShardBroadcaster 分片并行 + 分片失败独立补偿
v2.0：DagBuilder(jgrapht) DAG 编排 + DAG 状态表 + 运行态调分片数（任务定义由注解逐步放开到 DB 的动态入口，不破坏 v1 API）