easy-schedule 使用文档

嵌入式分布式定时任务组件，基于 db-scheduler（Apache-2.0）。 不部署独立服务，作为 SDK 嵌入 Spring Boot 应用，依赖 MySQL 做持久化。

1. 引入依赖

<dependency>
    <groupId>io.matrix</groupId>
    <artifactId>easy-schedule</artifactId>
    <version>${matrix.version}</version>
</dependency>

2. 准备数据库

执行 MySQL DDL（components/easy-schedule/sql/easy-schedule.sql）：

-- db-scheduler 引擎表（组件固定使用 sch_tasks 表名）
CREATE TABLE sch_tasks (
    task_name VARCHAR(100),
    task_instance VARCHAR(100),
    task_data LONGBLOB,
    execution_time DATETIME(6),
    picked TINYINT(1),
    picked_by VARCHAR(50),
    last_success DATETIME(6),
    last_failure DATETIME(6),
    consecutive_failures INT,
    last_heartbeat DATETIME(6),
    version BIGINT,
    priority SMALLINT,
    PRIMARY KEY (task_name, task_instance),
    INDEX idx_execution_time_picked (execution_time, picked)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

-- 动态 cron 配置表（运行时管理用）
CREATE TABLE sch_job_config (
    job_code     VARCHAR(96) PRIMARY KEY,
    cron         VARCHAR(64) NOT NULL COMMENT '6 位 cron 表达式',
    enabled      TINYINT(1) DEFAULT 1,
    shard_count  INT DEFAULT 1,
    payload_json JSON COMMENT '默认 payload（可选）',
    description  VARCHAR(256),
    owner        VARCHAR(64),
    created_at   DATETIME,
    updated_at   DATETIME,
    INDEX idx_enabled (enabled)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

sch_tasks 是 db-scheduler 引擎内部表，不要手动操作。sch_job_config 仅在启用动态 cron 时需要。

3. 配置

easy:
  schedule:
    enabled: true                          # 默认开启（需存在 DataSource）
    threads: 4                             # 引擎工作线程数
    polling-interval: 10s                  # 引擎轮询间隔
    enable-immediate-execution: true       # 调度到现在/过去时立即唤醒
    always-persist-timestamp-in-utc: true  # MySQL/MariaDB 建议开启
    dynamic-cron-enabled: true             # 是否启用动态 cron 运行时管理
    statistics-enabled: true               # 是否启用运行时内存统计

4. 定义任务

4.1 静态 cron（@Job 注解）

最简单的方式——在 Spring Bean 方法上加 @Job，组件启动时自动注册：

@Service
public class MyJobs {

    @Job(name = "daily-report", cron = "0 0 2 * * ?", desc = "每日凌晨2点生成报表")
    public void dailyReport() {
        // 直接写业务逻辑
        reportService.generate();
    }

    @Job(name = "data-sync", cron = "0 */5 * * * ?", shardCount = 3, desc = "每5分钟同步，3分片并行")
    public void dataSync(ExecutionContext ctx) {
        // 分片场景：每个节点只跑自己的分片
        int shardIndex = ctx.getShardIndex();  // 当前分片 (0 ~ shardCount-1)
        int shardTotal = ctx.getShardTotal();  // 总分片数
        syncService.syncShard(shardIndex, shardTotal);
    }

    @Job(name = "cleanup", cron = "0 0 3 * * ?", desc = "清理过期数据，带 payload")
    public void cleanup(CleanupPayload payload, ExecutionContext ctx) {
        cleanupService.cleanup(payload.getDays());
    }
}

@Job 参数说明：

方法签名支持 4 种形式：

void method()
void method(ExecutionContext ctx)
void method(Payload payload)           // Payload 会被 JSON 反序列化
void method(Payload payload, ExecutionContext ctx)

4.2 动态 cron（运行时注册）

通过 ScheduleManager 在运行时创建/修改/暂停 cron 任务：

@Autowired
private ScheduleManager scheduleManager;

// 注册一个动态 cron
String instanceId = scheduleManager.registerCron(
    "user-custom-report",     // jobCode，必须指向一个已有的 @Job 方法
    "0 0 8 * * ?",            // cron 表达式
    1,                        // shardCount
    null                      // 默认 payload（可选）
);

// 修改 cron 表达式
scheduleManager.updateCron("user-custom-report", "0 0 9 * * ?");

// 暂停（不再触发，但保留配置）
scheduleManager.pause("user-custom-report");

// 恢复
scheduleManager.resume("user-custom-report");

// 注销（删除配置 + 取消所有调度）
scheduleManager.unregister("user-custom-report");

// 列出所有动态 cron
List<DynamicJobConfig> all = scheduleManager.listDynamic();

注意：动态 cron 的 jobCode 必须指向一个已存在的 @Job 方法——组件在注册时会校验。

4.3 一次性任务

// 立即执行
scheduleManager.fireOnce("myJob", null, Instant.now());

// 延迟执行：5 分钟后
scheduleManager.fireOnce("myJob", payload, Instant.now().plus(5, ChronoUnit.MINUTES));

// 触发分片任务（等同于 cron 到期时的行为：广播 shardCount 个并行执行）
scheduleManager.trigger("data-sync");

5. 运行时 API

5.1 查询调度中/执行中的任务

// 查询即将执行的任务
List<ScheduledExecution<?>> scheduled = scheduleManager.listScheduled();

// 查询当前正在执行的任务（跨节点）
List<CurrentlyExecuting> running = scheduleManager.listExecuting();

5.2 取消/重调度

// 取消一个待执行的任务
scheduleManager.cancel(instanceId);

// 重新调度到新时间
scheduleManager.reschedule(instanceId, Instant.now().plus(10, ChronoUnit.MINUTES));

5.3 运行时统计

@Autowired
private SchedulerRuntime runtime;

// 调度器级别统计
SchedulerStats stats = runtime.getStatistics().getSchedulerStats();
stats.getTotalScheduled();   // 累计调度次数
stats.getTotalSuccess();     // 累计成功次数
stats.getTotalFailed();      // 累计失败次数
stats.getTotalDead();        // 累计超时死亡次数

// 单个 job 的统计
JobStats jobStats = runtime.getStatistics().getStats("myJob");
jobStats.getSuccessCount();
jobStats.getFailureCount();
jobStats.getLastSuccessAt();
jobStats.getLastError();
jobStats.getAvgDurationMs();
jobStats.getMaxDurationMs();

统计是进程内内存的，节点重启会清零。如需持久化统计，建议在业务层自行记录。

6. 集群行为

easy-schedule 是嵌入式组件，集群部署时：

• 所有节点连接同一个 MySQL，共享 sch_tasks 表
• 引擎通过行锁抢占（SELECT ... FOR UPDATE）保证同一时刻只有一个节点执行同一个任务
• 分片任务：cron 到期时由任意一个节点广播 shardCount 个一次性条目入库，各节点抢占并行执行
• 动态 cron 配置变更：改 sch_job_config 表后，所有节点的引擎在下一次轮询时会感知到（或调用 scheduler.triggerCheckForDueExecutions() 加速）

无需 Zookeeper、Redis 或任何中心化协调服务。

7. Cron 表达式

6 位标准格式：秒 分 时 日 月 周

8. 常见问题

Q: 启动时提示 “加载动态 cron 配置表失败（表可能不存在）”

A: 正常。如果不需要动态 cron 功能，可以设 easy.schedule.dynamic-cron-enabled=false 消除警告。

Q: 任务执行超时了怎么办？

A: db-scheduler 有心跳机制，超时的执行会标记为 DEAD 并重新调度。可通过 consecutiveFailures 观察。

Q: 如何手动触发一次调度检查？

A: scheduler.triggerCheckForDueExecutions() —— 立即扫描 sch_tasks 表，不等待轮询间隔。

Q: 引擎会阻塞应用启动吗？

A: 不会。引擎在独立线程池工作。启动过程只是把 @Job 注册进内存，然后后台线程开始轮询。