【功能新增】IoT：基于 Quartz 实现 IotSchedulerManager！为了兼容 boot 和 cloud！

YunaiV
1 parent 2109449a
Showing 6 changed files with 260 additions and 10 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/dataobject/rule/IotRuleSceneDO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/job/config/IotJobConfiguration.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/framework/job/core/IotSchedulerManager.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/IotRuleSceneServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneAction.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/rule/action/IotRuleSceneDataBridgeAction.java
@@ -60,8 +60,6 @@ public class IotRuleSceneDO extends TenantBaseDO {
     @TableField(typeHandler = JacksonTypeHandler.class)
     private List<TriggerConfig> triggers;
-    // TODO @芋艿：需要调研下 https://help.aliyun.com/zh/iot/user-guide/scene-orchestration-1?spm=a2c4g.11186623.help-menu-30520.d_2_4_5_0.45413908fxCSVa
-
     /**
      * 执行器数组
      */
@@ -97,7 +95,7 @@ public class IotRuleSceneDO extends TenantBaseDO {
         /**
          * 触发条件数组
          *
-         * 当 {@link #type} 为 {@link IotRuleSceneTriggerTypeEnum#DEVICE} 时，必填
+         * 必填：当 {@link #type} 为 {@link IotRuleSceneTriggerTypeEnum#DEVICE} 时
          * 条件与条件之间，是“或”的关系
          */
         private List<TriggerCondition> conditions;
@@ -105,7 +103,7 @@ public class IotRuleSceneDO extends TenantBaseDO {
         /**
          * CRON 表达式
          *
-         * 当 {@link #type} 为 {@link IotRuleSceneTriggerTypeEnum#TIMER} 时，必填
+         * 必填：当 {@link #type} 为 {@link IotRuleSceneTriggerTypeEnum#TIMER} 时
          */
         private String cronExpression;
@@ -185,15 +183,15 @@ public class IotRuleSceneDO extends TenantBaseDO {
         /**
          * 设备控制
          *
-         * 当 {@link #type} 为 {@link IotRuleSceneActionTypeEnum#DEVICE_CONTROL} 时，必填
+         * 必填：当 {@link #type} 为 {@link IotRuleSceneActionTypeEnum#DEVICE_CONTROL} 时
          */
         private ActionDeviceControl deviceControl;
         /**
          * 数据桥接编号
          *
-         * 当 {@link #type} 为 {@link IotRuleSceneActionTypeEnum#DATA_BRIDGE} 时，必填
-         * TODO 芋艿：关联
+         * 必填：当 {@link #type} 为 {@link IotRuleSceneActionTypeEnum#DATA_BRIDGE} 时
+         * 关联：{@link IotDataBridgeDO#getId()}
          */
         private Long dataBridgeId;
+package cn.iocoder.yudao.module.iot.framework.job.config;
+
+import cn.iocoder.yudao.module.iot.framework.job.core.IotSchedulerManager;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.sql.DataSource;
+
+/**
+ * IoT 模块的 Job 自动配置类
+ *
+ * @author 芋道源码
+ */
+@Configuration
+public class IotJobConfiguration {
+
+    @Bean(initMethod = "start", destroyMethod = "stop")
+    public IotSchedulerManager iotSchedulerManager(DataSource dataSource,
+                                                   ApplicationContext applicationContext) {
+        return new IotSchedulerManager(dataSource, applicationContext);
+    }
+
+}
+package cn.iocoder.yudao.module.iot.framework.job.core;
+
+import cn.iocoder.yudao.framework.quartz.core.enums.JobDataKeyEnum;
+import lombok.extern.slf4j.Slf4j;
+import org.quartz.*;
+import org.springframework.context.ApplicationContext;
+import org.springframework.scheduling.quartz.SchedulerFactoryBean;
+import org.springframework.scheduling.quartz.SpringBeanJobFactory;
+
+import javax.sql.DataSource;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * IoT 模块的 Scheduler 管理类，基于 Quartz 实现
+ *
+ * 疑问：为什么 IoT 模块不复用全局的 SchedulerManager 呢？
+ * 回复：yudao-cloud 项目，使用的是 XXL-Job 作为调度中心，无法动态添加任务。
+ *
+ * @author 芋道源码
+ */
+@Slf4j
+public class IotSchedulerManager {
+
+    private static final String SCHEDULER_NAME = "iotScheduler";
+
+    private final SchedulerFactoryBean schedulerFactoryBean;
+
+    private Scheduler scheduler;
+
+    public IotSchedulerManager(DataSource dataSource,
+                               ApplicationContext applicationContext) {
+        // 1. 参考 SchedulerFactoryBean 类
+        SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
+        SpringBeanJobFactory jobFactory = new SpringBeanJobFactory();
+        jobFactory.setApplicationContext(applicationContext);
+        schedulerFactoryBean.setJobFactory(jobFactory);
+        schedulerFactoryBean.setAutoStartup(true);
+        schedulerFactoryBean.setSchedulerName(SCHEDULER_NAME);
+        schedulerFactoryBean.setDataSource(dataSource);
+        schedulerFactoryBean.setWaitForJobsToCompleteOnShutdown(true);
+        Properties properties = new Properties();
+        schedulerFactoryBean.setQuartzProperties(properties);
+        // 2. 参考 application-local.yaml 配置文件
+        // 2.1 Scheduler 相关配置
+        properties.put("org.quartz.scheduler.instanceName", SCHEDULER_NAME);
+        properties.put("org.quartz.scheduler.instanceId", "AUTO");
+        // 2.2 JobStore 相关配置
+        properties.put("org.quartz.jobStore.class", "org.springframework.scheduling.quartz.LocalDataSourceJobStore");
+        properties.put("org.quartz.jobStore.isClustered", "true");
+        properties.put("org.quartz.jobStore.clusterCheckinInterval", "15000");
+        properties.put("org.quartz.jobStore.misfireThreshold", "60000");
+        // 2.3 线程池相关配置
+        properties.put("org.quartz.threadPool.threadCount", "25");
+        properties.put("org.quartz.threadPool.threadPriority", "5");
+        properties.put("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
+        this.schedulerFactoryBean = schedulerFactoryBean;
+    }
+
+    public void start() throws Exception {
+        log.info("[start][Scheduler 初始化开始]");
+        // 初始化
+        schedulerFactoryBean.afterPropertiesSet();
+        schedulerFactoryBean.start();
+        // 获得 Scheduler 对象
+        this.scheduler = schedulerFactoryBean.getScheduler();
+        log.info("[start][Scheduler 初始化完成]");
+    }
+
+    public void stop() {
+        log.info("[stop][Scheduler 关闭开始]");
+        schedulerFactoryBean.stop();
+        this.scheduler = null;
+        log.info("[stop][Scheduler 关闭完成]");
+    }
+
+    // ========== 参考 SchedulerManager 实现 ==========
+
+    /**
+     * 添加或更新 Job 到 Quartz 中
+     *
+     * @param jobClass 任务处理器的类
+     * @param jobHandlerName 任务处理器的名字
+     * @param cronExpression CRON 表达式
+     * @param jobDataMap 任务数据
+     * @throws SchedulerException 添加异常
+     */
+    public void addOrUpdateJob(Class <? extends Job> jobClass, String jobHandlerName,
+                               String cronExpression, Map<String, Object> jobDataMap)
+            throws SchedulerException {
+        if (scheduler.checkExists(new JobKey(jobHandlerName))) {
+            this.updateJob(jobHandlerName, cronExpression);
+        } else {
+            this.addJob(jobClass, jobHandlerName, cronExpression, jobDataMap);
+        }
+    }
+
+    /**
+     * 添加 Job 到 Quartz 中
+     *
+     * @param jobClass 任务处理器的类
+     * @param jobHandlerName 任务处理器的名字
+     * @param cronExpression CRON 表达式
+     * @param jobDataMap 任务数据
+     * @throws SchedulerException 添加异常
+     */
+    public void addJob(Class <? extends Job> jobClass, String jobHandlerName,
+                       String cronExpression, Map<String, Object> jobDataMap)
+            throws SchedulerException {
+        // 创建 JobDetail 对象
+        JobDetail jobDetail = JobBuilder.newJob(jobClass)
+                .usingJobData(new JobDataMap(jobDataMap))
+                .withIdentity(jobHandlerName).build();
+        // 创建 Trigger 对象
+        Trigger trigger = this.buildTrigger(jobHandlerName, cronExpression);
+        // 新增 Job 调度
+        scheduler.scheduleJob(jobDetail, trigger);
+    }
+
+    /**
+     * 更新 Job 到 Quartz
+     *
+     * @param jobHandlerName 任务处理器的名字
+     * @param cronExpression CRON 表达式
+     * @throws SchedulerException 更新异常
+     */
+    public void updateJob(String jobHandlerName, String cronExpression)
+            throws SchedulerException {
+        // 创建新 Trigger 对象
+        Trigger newTrigger = this.buildTrigger(jobHandlerName, cronExpression);
+        // 修改调度
+        scheduler.rescheduleJob(new TriggerKey(jobHandlerName), newTrigger);
+    }
+
+    /**
+     * 删除 Quartz 中的 Job
+     *
+     * @param jobHandlerName 任务处理器的名字
+     * @throws SchedulerException 删除异常
+     */
+    public void deleteJob(String jobHandlerName) throws SchedulerException {
+        // 暂停 Trigger 对象
+        scheduler.pauseTrigger(new TriggerKey(jobHandlerName));
+        // 取消并删除 Job 调度
+        scheduler.unscheduleJob(new TriggerKey(jobHandlerName));
+        scheduler.deleteJob(new JobKey(jobHandlerName));
+    }
+
+    /**
+     * 暂停 Quartz 中的 Job
+     *
+     * @param jobHandlerName 任务处理器的名字
+     * @throws SchedulerException 暂停异常
+     */
+    public void pauseJob(String jobHandlerName) throws SchedulerException {
+        scheduler.pauseJob(new JobKey(jobHandlerName));
+    }
+
+    /**
+     * 启动 Quartz 中的 Job
+     *
+     * @param jobHandlerName 任务处理器的名字
+     * @throws SchedulerException 启动异常
+     */
+    public void resumeJob(String jobHandlerName) throws SchedulerException {
+        scheduler.resumeJob(new JobKey(jobHandlerName));
+        scheduler.resumeTrigger(new TriggerKey(jobHandlerName));
+    }
+
+    /**
+     * 立即触发一次 Quartz 中的 Job
+     *
+     * @param jobHandlerName 任务处理器的名字
+     * @throws SchedulerException 触发异常
+     */
+    public void triggerJob(String jobHandlerName)
+            throws SchedulerException {
+        // 触发任务
+        JobDataMap data = new JobDataMap();
+        data.put(JobDataKeyEnum.JOB_HANDLER_NAME.name(), jobHandlerName);
+        scheduler.triggerJob(new JobKey(jobHandlerName), data);
+    }
+
+    private Trigger buildTrigger(String jobHandlerName, String cronExpression) {
+        return TriggerBuilder.newTrigger()
+                .withIdentity(jobHandlerName)
+                .withSchedule(CronScheduleBuilder.cronSchedule(cronExpression))
+                .build();
+    }
+
+}
@@ -23,6 +23,11 @@ import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.service.rule.action.IotRuleSceneAction;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.TriggerKey;
+import org.quartz.impl.StdSchedulerFactory;
 import org.springframework.stereotype.Service;
 import org.springframework.validation.annotation.Validated;
@@ -330,4 +335,28 @@ public class IotRuleSceneServiceImpl implements IotRuleSceneService {
         });
     }
+    // TODO @芋艿：测试思路代码，记得删除！！！
+    // 1. Job 类：IotRuleSceneJob
+    // 2. 参数：id
+    // 3. jobHandlerName：IotRuleSceneJob + id
+
+    // 新增：addJob
+    // 修改：不存在 addJob、存在 updateJob
+    // 有 + 禁用：1）存在、停止；2）不存在：不处理；TODO 测试：直接暂停，是否可以？？？（结论：可以）
+    // 有 + 开启：1）存在，更新；2）不存在，新增；
+    // 无 + 禁用、开启：1）存在，删除；TODO 测试：直接删除？？？（结论：可以）
+
+    public static void main2(String[] args) throws SchedulerException {
+//        System.out.println(QuartzJobBean.class);
+        Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
+        scheduler.start();
+
+        String jobHandlerName = "123";
+        // 暂停 Trigger 对象
+        scheduler.pauseTrigger(new TriggerKey(jobHandlerName));
+        // 取消并删除 Job 调度
+        scheduler.unscheduleJob(new TriggerKey(jobHandlerName));
+        scheduler.deleteJob(new JobKey(jobHandlerName));
+    }
+
 }
@@ -4,6 +4,8 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.rule.IotRuleSceneDO;
 import cn.iocoder.yudao.module.iot.enums.rule.IotRuleSceneActionTypeEnum;
 import cn.iocoder.yudao.module.iot.mq.message.IotDeviceMessage;
+import javax.annotation.Nullable;
+
 /**
  * IOT 规则场景的场景执行器接口
  *
@@ -14,10 +16,12 @@ public interface IotRuleSceneAction {
     /**
      * 执行场景
      *
-     * @param message 消息
+     * @param message 消息，允许空
+     *                1. 空的情况：定时触发
+     *                2. 非空的情况：设备触发
      * @param config 配置
      */
-    void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config);
+    void execute(@Nullable IotDeviceMessage message, IotRuleSceneDO.ActionConfig config);
     /**
      * 获得类型
@@ -43,7 +43,11 @@ public class IotRuleSceneDataBridgeAction implements IotRuleSceneAction {
     @Override
     public void execute(IotDeviceMessage message, IotRuleSceneDO.ActionConfig config) {
-        // 1. 获得数据桥梁
+        // 1.1 如果消息为空，直接返回
+        if (message == null) {
+            return;
+        }
+        // 1.2 获得数据桥梁
         Assert.notNull(config.getDataBridgeId(), "数据桥梁编号不能为空");
         IotDataBridgeDO dataBridge = dataBridgeService.getIotDataBridge(config.getDataBridgeId());
         if (dataBridge == null || dataBridge.getConfig() == null) {