feat: 切换定时任务至分布式调度底座

- 以执行账本和有界 Worker 承载重负载任务与故障接管 - 接入统一调度 Starter 并增加 MySQL 迁移、指标和回归测试
2026-08-31 14:57:08 +08:00
parent 17ef189862
commit 8c174e5c02
66 changed files with 5348 additions and 545 deletions
--- a/easyflow-starter/easyflow-starter-all/src/main/java/tech/easyflow/starter/MainApplication.java
+++ b/easyflow-starter/easyflow-starter-all/src/main/java/tech/easyflow/starter/MainApplication.java
@@ -3,12 +3,16 @@ package tech.easyflow.starter;
 import org.dromara.x.file.storage.spring.EnableFileStorage;
 import org.springframework.boot.actuate.autoconfigure.elasticsearch.ElasticsearchRestHealthContributorAutoConfiguration;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.quartz.QuartzAutoConfiguration;
 import tech.easyflow.common.spring.BaseApp;

 /**
 * EasyFlow 启动入口。
 */
-@SpringBootApplication(exclude = ElasticsearchRestHealthContributorAutoConfiguration.class)
+@SpringBootApplication(exclude = {
+        ElasticsearchRestHealthContributorAutoConfiguration.class,
+        QuartzAutoConfiguration.class
+})
@EnableFileStorage
 public class MainApplication extends BaseApp {

--- a/easyflow-starter/easyflow-starter-all/src/main/java/tech/easyflow/starter/MybatisConfig.java
+++ b/easyflow-starter/easyflow-starter-all/src/main/java/tech/easyflow/starter/MybatisConfig.java
@@ -1,9 +1,8 @@
 package tech.easyflow.starter;

 import com.mybatisflex.core.FlexGlobalConfig;
+import com.mybatisflex.core.audit.AuditMessage;
 import com.mybatisflex.core.audit.AuditManager;
-import com.mybatisflex.core.audit.ConsoleMessageCollector;
-import com.mybatisflex.core.audit.MessageCollector;
 import com.mybatisflex.core.tenant.TenantManager;
 import com.mybatisflex.spring.boot.MyBatisFlexCustomizer;
 import cn.dev33.satoken.stp.StpUtil;
@@ -11,18 +10,27 @@ import cn.dev33.satoken.exception.NotWebContextException;
 import cn.dev33.satoken.exception.SaTokenContextException;
 import cn.dev33.satoken.context.SaHolder;
 import java.math.BigInteger;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Configuration;
 import tech.easyflow.common.entity.LoginAccount;
 import tech.easyflow.common.satoken.util.SaTokenUtil;
+import tech.easyflow.job.execution.SysJobExecutionContextHolder;

@Configuration
 public class MybatisConfig implements MyBatisFlexCustomizer {

+    private static final Logger log = LoggerFactory.getLogger(MybatisConfig.class);
+    private static final int MAX_AUDIT_SQL_LENGTH = 2_000;
+
+    @Value("${easyflow.mybatis.sql-audit-enabled:false}")
+    private boolean sqlAuditEnabled;

    @Override
    public void customize(FlexGlobalConfig flexGlobalConfig) {
-        //开启审计功能
-        AuditManager.setAuditEnable(true);
+        // SQL 审计会同步格式化并输出每条语句，高频 Worker 轮询下必须显式开启。
+        AuditManager.setAuditEnable(sqlAuditEnabled);

        // 统一使用标准 0/1 逻辑删除语义。
        flexGlobalConfig.setNormalValueOfLogicDelete(0);
@@ -34,9 +42,25 @@ public class MybatisConfig implements MyBatisFlexCustomizer {
        //取消控制台的 Banner 打印
        flexGlobalConfig.setPrintBanner(false);

-        //设置 SQL 审计收集器
-        MessageCollector collector = new ConsoleMessageCollector();
-        AuditManager.setMessageCollector(collector);
+        if (sqlAuditEnabled) {
+            AuditManager.setMessageCollector(MybatisConfig::collectSafeAuditMessage);
+        }
+    }
+
+    /**
+     * 审计日志只记录参数化 SQL 模板，绝不读取 AuditMessage 中的参数或完整 SQL。
+     */
+    private static void collectSafeAuditMessage(AuditMessage message) {
+        log.info("SQL audit: statement={}, elapsedMs={}, rows={}, sql={}",
+            message.getStmtId(), message.getElapsedTime(), message.getQueryCount(),
+            compactSqlTemplate(message.getQuery()));
+    }
+
+    private static String compactSqlTemplate(String sql) {
+        if (sql == null) return null;
+        String compact = sql.replaceAll("\\s+", " ").trim();
+        return compact.length() <= MAX_AUDIT_SQL_LENGTH
+            ? compact : compact.substring(0, MAX_AUDIT_SQL_LENGTH) + "...";
    }

    /**
@@ -45,6 +69,12 @@ public class MybatisConfig implements MyBatisFlexCustomizer {
     * @return 当前租户 ID 数组；没有请求登录上下文时返回 {@code null}
     */
    private Object[] currentTenantIds() {
+        BigInteger scheduledTenantId = SysJobExecutionContextHolder.current()
+                .map(context -> context.tenantId())
+                .orElse(null);
+        if (scheduledTenantId != null) {
+            return new Object[]{scheduledTenantId};
+        }
        try {
            if (!SaHolder.getContext().isValid() || !StpUtil.isLogin()) {
                return null;
--- a/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml
+++ b/easyflow-starter/easyflow-starter-all/src/main/resources/application.yml
@@ -25,7 +25,8 @@ spring:
    username: root
    password: root
    hikari:
-      maximum-pool-size: 12
+      # 调度管理 Saga 会同时占用业务事务和 Quartz JobStore 连接；容量由启动校验守护。
+      maximum-pool-size: 24
      minimum-idle: 2
      connection-timeout: 5000
      validation-timeout: 3000
@@ -59,34 +60,65 @@ spring:
    # 静态资源路径，用于访问本地文件
    # ！！！ 注意，这里要和下面的 easyflow.storage.local.prefix 后面的路径 /attachment 保持一致！
    static-path-pattern: /attachment/**
-  # quartz 相关配置
-  quartz:
-    startup-delay: 1
-    job-store-type: jdbc
-    jdbc:
-      platform: mysql
-      initialize-schema: never
-    properties:
-      org:
-        quartz:
-          scheduler:
-            instanceName: easyflowScheduler
-            instanceId: AUTO
-          jobStore:
-            misfireThreshold: 1000
-            isClustered: true
-            clusterCheckinInterval: 15000
-            # 如果数据库大小写敏感，可将ddl里的相关表名改为大写
-            tablePrefix: TB_QRTZ_
-            driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
-          threadPool:
-            threadCount: 8
-            threadPriority: 5
  threads:
    virtual:
      enabled: true

+easy-agents:
+  scheduler:
+    enabled: true
+    data-source-bean-name: dataSource
+    quartz:
+      scheduler-name: easyflowScheduler
+      instance-id: AUTO
+      table-prefix: TB_QRTZ_
+      driver-delegate-class: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
+      clustered: true
+      thread-count: 8
+      thread-priority: 5
+      # 短 Handler 可按可用线程批量获取同一计划时刻的 Trigger；批量模式由底座强制在锁内获取。
+      batch-trigger-acquisition-max-count: 8
+      # 保持 0，避免下一计划时刻的任务被同一批次提前执行。
+      batch-trigger-acquisition-fire-ahead-time-window-millis: 0
+      cluster-checkin-interval-millis: 15000
+      misfire-threshold-millis: 60000
+      wait-for-jobs-to-complete-on-shutdown: true
+      shutdown-wait-timeout-millis: 25000
+      validate-schema: true
+  # easy-agents 文档解析统一配置
+  document:
+    ocr:
+      provider: mineru
+      mineru:
+        # 统一文档解析桥接层直接复用 easy-agents 的 provider 配置，不在 easyflow 再复制一套配置体系
+        base-url: https://ontoweb.wust.edu.cn/mineru-api
+        submit-timeout-ms: 120000
+        default-lang-list:
+          - ch
+
 easyflow:
+  mybatis:
+    # SQL 审计会同步输出脱敏后的参数化模板，仅在受控排查时临时开启。
+    sql-audit-enabled: ${EASYFLOW_MYBATIS_SQL_AUDIT_ENABLED:false}
+  job:
+    timezone: Asia/Shanghai
+    execution:
+      enabled: true
+      worker-count: 4
+      # 管理命令会同时占用业务事务与 Quartz JobStore 连接，需为连接池保留余量。
+      management-command-concurrency: 4
+      business-connection-reserve: 4
+      poll-interval: 500ms
+      lease-duration: 2m
+      heartbeat-interval: 30s
+      retry-backoff: 5s
+      shutdown-wait-timeout: 30s
+      infrastructure-retry-limit: 16
+      registration-max-attempts: 3
+      registration-retry-delay: 100ms
+      # 0 表示账本登记持续失败时保留同一次 Quartz fire，直至数据库恢复。
+      registration-quartz-refire-limit: 0
+      registration-quartz-refire-delay: 250ms
  datacenter:
    # 数据源凭据使用独立部署密钥，禁止与仓库内其他固定密钥复用。
    credential-key: cA0ldVrBRBWJlWN7BvaAGXmxH1+wCpoSE0y/iSudQsQ=
@@ -289,18 +321,6 @@ dromara:
        bucket-name: easyflow-agent-artifacts
        base-path: published

-# easy-agents 文档解析统一配置
-easy-agents:
-  document:
-    ocr:
-      provider: mineru
-      mineru:
-        # 统一文档解析桥接层直接复用 easy-agents 的 provider 配置，不在 easyflow 再复制一套配置体系
-        base-url: https://ontoweb.wust.edu.cn/mineru-api
-        submit-timeout-ms: 120000
-        default-lang-list:
-          - ch
-
 # 自定义节点相关配置
 node:
  # 文件内容提取节点，默认使用简单文档读取器，可自行实现 ReadDocService
--- a/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V62__mysql_sys_job_execution_ledger.sql
+++ b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V62__mysql_sys_job_execution_ledger.sql
@@ -0,0 +1,33 @@
+ALTER TABLE `tb_sys_job`
+    ADD COLUMN `schedule_generation` bigint NOT NULL DEFAULT 0 COMMENT '调度定义代际' AFTER `status`;
+
+ALTER TABLE `tb_sys_job_log`
+    ADD COLUMN `execution_key` char(64) CHARACTER SET ascii COLLATE ascii_bin NOT NULL COMMENT '执行幂等键SHA-256' AFTER `id`,
+    ADD COLUMN `job_generation` bigint NOT NULL COMMENT '触发所属任务代际' AFTER `job_id`,
+    ADD COLUMN `tenant_id` bigint UNSIGNED NOT NULL COMMENT '租户ID' AFTER `job_id`,
+    ADD COLUMN `dept_id` bigint UNSIGNED NOT NULL COMMENT '部门ID' AFTER `tenant_id`,
+    ADD COLUMN `job_type` int NOT NULL COMMENT '任务类型快照' AFTER `job_name`,
+    ADD COLUMN `job_options` text CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL COMMENT '任务扩展配置快照' AFTER `job_params`,
+    ADD COLUMN `allow_concurrent` int NOT NULL DEFAULT 0 COMMENT '是否允许并发执行' AFTER `job_options`,
+    ADD COLUMN `trigger_source` varchar(16) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '触发来源' AFTER `allow_concurrent`,
+    ADD COLUMN `invocation_id` varchar(190) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL COMMENT '立即触发调用标识' AFTER `trigger_source`,
+    ADD COLUMN `scheduled_fire_time` datetime(3) NOT NULL COMMENT '计划触发时间' AFTER `trigger_source`,
+    ADD COLUMN `actual_fire_time` datetime(3) NOT NULL COMMENT '实际触发时间' AFTER `scheduled_fire_time`,
+    ADD COLUMN `fire_instance_id` varchar(190) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT 'Quartz物理触发实例' AFTER `actual_fire_time`,
+    ADD COLUMN `recovering` tinyint NOT NULL DEFAULT 0 COMMENT '是否为Quartz故障恢复' AFTER `fire_instance_id`,
+    ADD COLUMN `lease_owner` varchar(190) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL COMMENT '执行节点' AFTER `recovering`,
+    ADD COLUMN `execution_token` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL COMMENT '本轮执行令牌' AFTER `lease_owner`,
+    ADD COLUMN `lease_until` datetime(3) NULL COMMENT '租约到期时间' AFTER `execution_token`,
+    ADD COLUMN `heartbeat_time` datetime(3) NULL COMMENT '最近续租时间' AFTER `lease_until`,
+    ADD COLUMN `attempt_count` int NOT NULL DEFAULT 0 COMMENT '基础设施恢复次数' AFTER `heartbeat_time`,
+    ADD COLUMN `next_retry_time` datetime(3) NULL COMMENT '下次恢复时间' AFTER `attempt_count`,
+    ADD COLUMN `version` bigint NOT NULL DEFAULT 0 COMMENT '状态版本' AFTER `next_retry_time`,
+    MODIFY COLUMN `status` int NOT NULL DEFAULT 2 COMMENT '0失败 1成功 2等待执行 3执行中 4需人工处理 5已取消',
+    MODIFY COLUMN `start_time` datetime(3) NULL COMMENT '开始时间',
+    MODIFY COLUMN `end_time` datetime(3) NULL COMMENT '结束时间',
+    ADD UNIQUE KEY `uk_sys_job_log_execution_key` (`execution_key`),
+    ADD KEY `idx_sys_job_log_pending_claim` (`status`, `next_retry_time`, `id`),
+    ADD KEY `idx_sys_job_log_expired_claim` (`status`, `lease_until`, `id`),
+    ADD KEY `idx_sys_job_log_job_active` (`job_id`, `status`, `lease_until`),
+    ADD KEY `idx_sys_job_log_job_created` (`job_id`, `created`),
+    ADD KEY `idx_sys_job_log_tenant_id` (`tenant_id`, `id`);
--- a/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V63__mysql_sys_job_log_time_query_indexes.sql
+++ b/easyflow-starter/easyflow-starter-all/src/main/resources/db/migration/mysql/V63__mysql_sys_job_log_time_query_indexes.sql
@@ -0,0 +1,3 @@
+ALTER TABLE `tb_sys_job_log`
+    ADD KEY `idx_sys_job_log_job_scheduled` (`job_id`, `scheduled_fire_time`, `id`),
+    ADD KEY `idx_sys_job_log_job_actual` (`job_id`, `actual_fire_time`, `id`);