调度模块对接文档

以预警模块为例，说明如何对接调度模块。

一、对接前提

调度模块基于 PowerJob 实现，对接前需确认：

• powerjob.worker.enabled = true（生产环境必须开启）
• PowerJob Server 已部署并可访问
• 应用已在 PowerJob Server 注册（appName: sys）

二、对接步骤（预警模块示例）

2.1 预警模块需求

预警模块需要定时扫描业务数据，根据策略配置判断是否触发预警，并通过消息渠道推送。

定时逻辑:

• 按策略配置的 interval_minutes 周期扫描
• 不同策略可配置不同的扫描频率
• 遵守 prohibit_repeated_cycle（禁止重复周期），避免同一数据重复预警

2.2 数据库关系

sys_schedule                         sys_warn_strategy
┌──────────────────────┐             ┌───────────────────────────┐
│ id (PK)              │◄────────────│ schedule_id (FK)          │
│ code                 │             │ id (PK)                   │
│ name                 │             │ node_id                   │
│ cron_expression      │             │ interval_minutes          │
│ class_name           │             │ is_enabled                │
│ job_id               │             │ prohibit_repeated_cycle   │
│ param                │             │ ...                       │
└──────────────────────┘             └───────────────────────────┘

sys_warn_strategy.schedule_id 关联 sys_schedule.id，一个调度可被多个策略复用。

2.3 实现 IScheduleProcessor

package com.sie.imom.sys.warn.processor;

/**
 * 预警扫描处理器
 *
 * @author luozhanfeng
 */
@Component
public class WarnScanProcessor implements IScheduleProcessor {

    @Autowired
    private WarnStrategyService warnStrategyService;

    @Autowired
    private WarnPushService warnPushService;

    @Override
    public String execute(String param) {
        // param 是 sys_schedule.param 的 JSON，可包含额外配置
        // 此处从策略表读取配置

        // 1. 查询所有启用的预警策略（按 schedule_id 筛选）
        List<WarnStrategy> strategies = warnStrategyService.listEnabledByScheduleId(getScheduleId());

        int totalChecked = 0;
        int totalWarned = 0;

        for (WarnStrategy strategy : strategies) {
            // 2. 根据节点配置查询业务数据
            List<WarnCheckResult> results = warnStrategyService.check(strategy);

            // 3. 判断预警等级（operator + value / value_next 区间判断）
            for (WarnCheckResult result : results) {
                WarnLevel level = warnStrategyService.evaluateLevel(strategy, result);
                if (level == null) {
                    continue; // 未达到预警阈值
                }

                // 4. 禁止重复周期检查
                if (warnStrategyService.isInProhibitPeriod(strategy, result)) {
                    continue;
                }

                // 5. 执行推送（邮件、站内信等）
                warnPushService.push(strategy, level, result);
                totalWarned++;
            }
            totalChecked += results.size();
        }

        return String.format("扫描完成: 检查%d条, 触发预警%d条", totalChecked, totalWarned);
    }
}

2.4 创建调度配置

调用调度模块 API 创建调度记录：

POST /sys/v1/schedule
{
    "code": "warnScan",
    "name": "预警定时扫描",
    "cronExpression": "0 */5 * * * ?",          // 每5分钟执行
    "className": "com.sie.imom.sys.warn.processor.WarnScanProcessor",
    "param": "{}",
    "description": "扫描所有启用的预警策略，判断是否触发预警",
    "isEnabled": "1"
}

创建成功后：

• sys_schedule 插入一条记录，job_id 由 PowerJob 自动回填
• PowerJob Server 按 cron 表达式定时触发 WarnScanProcessor.execute()
• 策略表中的 schedule_id 指向该调度记录

2.5 关联策略到调度

在预警策略创建/编辑时，将策略关联到调度：

POST /sys/v1/warn/strategy
{
    "scheduleId": 123456789,              // 关联调度 ID
    "nodeId": 1001,
    "nodeCode": "temperature",
    "isEnabled": "1",
    "intervalMinutes": 30,
    "prohibitRepeatedCycle": 24,
    "repeatedWarningType": "NEW",
    "levels": [
        {
            "level": "WARNING",
            "operator": "GT",
            "value": 80,
            "pushes": [{"templateId": 100}]
        }
    ]
}

2.6 执行日志

每次调度触发后，自动记录 sys_schedule_log：

sys_schedule_log
┌──────────────────────────────────────────────────┐
│ schedule_id: 123456789                           │
│ job_id: "powerjob-xxx-123"                       │
│ status: SUCCESS                                  │
│ result: "扫描完成: 检查150条, 触发预警3条"          │
│ start_time: 2026-06-16 10:00:00                  │
│ end_time: 2026-06-16 10:00:02                    │
│ duration: 2345                                   │
└──────────────────────────────────────────────────┘

预警模块自身也有 sys_warn_log + sys_warn_log_detail 记录推送结果，与调度日志形成两级日志体系。

三、完整数据流

1. 配置阶段
   用户 → ScheduleController → sys_schedule (调度配置)
   用户 → WarnStrategyController → sys_warn_strategy (关联 schedule_id)

2. 触发阶段
   PowerJob Server → PowerJob Worker → ScheduleProcessorDispatcher
       → 查 sys_schedule → 反射 IScheduleProcessor Bean
       → WarnScanProcessor.execute()
           → warnStrategyService.check()        → 业务数据查询
           → warnStrategyService.evaluateLevel() → 等级判断
           → warnPushService.push()             → 消息推送
               → sys_warn_log (推送日志)
               → sys_warn_log_detail (收件人明细)
       → 写 sys_schedule_log (调度日志)

四、注意事项

4.1 处理器类名正确性

sys_schedule.class_name 必须是 Spring Bean 的全限定类名，且该类实现了 IScheduleProcessor 接口。类名错误会导致反射失败，任务执行报错。

验证方式：手动触发一次调度，确认日志正常。

4.2 分布式执行

PowerJob 是分布式调度引擎，同一任务同一时刻只会在一个 Worker 节点上执行，但仍建议：

• 处理器逻辑保证幂等性（重复执行不产生副作用）
• 避免在处理器中写死单机资源（如本地文件路径）

4.3 参数设计

sys_schedule.param 是 JSON 格式的默认参数，传给处理器的 execute(param) 方法。按需使用：

• 简单场景：不需要参数，传 "{}"
• 需要区分场景：传 {"type": "daily"} 等
• 动态参数：通过 sys_schedule_log.run_param 在手动触发时传递独立参数

4.4 调度与策略的关系

• 一对多：一个调度可服务多个预警策略（通过 schedule_id 关联）
• 按需调整 cron：如果策略要求不同的扫描频率，可创建多个调度
• 策略开关不影响调度：关闭策略只是 sys_warn_strategy.is_enabled=0，调度照常执行，处理器内部跳过即可

4.5 异常处理

处理器内部必须捕获所有异常，不能向上抛：

@Override
public String execute(String param) {
    try {
        // 业务逻辑
        return "成功";
    } catch (Exception e) {
        log.error("预警扫描异常", e);
        return "失败: " + e.getMessage();  // 不要 throw
    }
}

ScheduleProcessorDispatcher 会捕获异常并写入 sys_schedule_log.error_message，但如果处理器内部吞掉异常可以返回更友好的错误信息。

4.6 禁止重复预警

预警模块通过 sys_warn_strategy.prohibit_repeated_cycle（小时）控制重复预警。实现方式：

• 推送前查询 sys_warn_log，判断同一策略+同一节点是否在禁止周期内已有预警记录
• 如果存在且在周期内，跳过本次推送

4.7 邮件推送渠道

预警推送通过 EmailUtil（位于 com.sie.imom.sys.util，当前为注释状态）发送邮件。对接时需：

• 配置 SMTP 参数（服务器、端口、账号、授权码）
• 预警模板表（sys_warn_template）存储消息模板，支持变量替换
• sys_warn_channel 管理可用渠道

4.8 应用启动初始化

WarnInitService.init()（位于 RunController 中，当前为注释状态）负责初始化预警枚举和默认调度配置。对接完成后需：

• 实现 WarnInitService，注册默认调度记录和预警分类
• 在 SysRunner 或 RunController 中调用初始化

4.9 开发环境建议

• 设置 powerjob.worker.enabled = false，避免本地开发环境连接远程 PowerJob Server
• 可手动调用 ScheduleController 的"手动触发"接口来测试处理器
• 调度日志保留 30 天，定期清理

五、预警模块数据模型速查