优化定时调度

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-api/src/main/java/org/hswebframework/web/service/schedule/ScheduleJobExecutor.java

@@ -3,8 +3,6 @@ package org.hswebframework.web.service.schedule;
 import java.util.Map;
 
 /**
- * TODO 完成注释
- *
  * @author zhouhao
  */
 public interface ScheduleJobExecutor {

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/pom.xml

@@ -22,5 +22,10 @@
             <artifactId>hsweb-system-schedule-service-api</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-message-api</artifactId>
+            <version>${project.version}</version>
+        </dependency>
     </dependencies>
 </project>

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/DefaultScriptScheduleJobExecutor.java

@@ -15,11 +15,17 @@ import java.util.Map;
 /**
  * @author zhouhao
  */
-@Service
 public class DefaultScriptScheduleJobExecutor implements ScheduleJobExecutor {
 
     private ScheduleJobService scheduleJobService;
 
+    public DefaultScriptScheduleJobExecutor() {
+    }
+
+    public DefaultScriptScheduleJobExecutor(ScheduleJobService scheduleJobService) {
+        this.scheduleJobService = scheduleJobService;
+    }
+
     @Autowired
     public void setScheduleJobService(ScheduleJobService scheduleJobService) {
         this.scheduleJobService = scheduleJobService;

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/DynamicJobFactory.java

@@ -11,8 +11,6 @@ import org.springframework.beans.factory.annotation.Autowired;
 import java.util.Map;
 
 /**
- * TODO 完成注释
- *
  * @author zhouhao
  */
 public class DynamicJobFactory implements JobFactory {

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/SmartScheduleTriggerBuilder.java

@@ -8,8 +8,6 @@ import org.quartz.spi.MutableTrigger;
 import org.springframework.stereotype.Service;
 
 /**
- * TODO 完成注释
- *
  * @author zhouhao
  */
 @Service

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/cluster/JobExecuteResultMessage.java

@@ -0,0 +1,22 @@
+package org.hswebframework.web.service.schedule.simple.cluster;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.hswebframework.web.message.Message;
+
+/**
+ * @author zhouhao
+ * @since 3.0
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class JobExecuteResultMessage implements Message {
+    private String executeId;
+
+    private boolean success;
+
+    private Object  result;
+
+}

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/cluster/JobExecutorMessage.java

@@ -0,0 +1,24 @@
+package org.hswebframework.web.service.schedule.simple.cluster;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.hswebframework.web.message.Message;
+
+import java.util.Map;
+
+/**
+ * @author zhouhao
+ * @since 3.0
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class JobExecutorMessage implements Message {
+
+    private String executeId;
+
+    private String jobId;
+
+    private Map<String, Object> parameters;
+}

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-service/hsweb-system-schedule-service-simple/src/main/java/org/hswebframework/web/service/schedule/simple/cluster/MessagerScheduleJobExecutor.java

@@ -0,0 +1,137 @@
+package org.hswebframework.web.service.schedule.simple.cluster;
+
+import lombok.extern.slf4j.Slf4j;
+import org.hswebframework.web.entity.schedule.ScheduleJobEntity;
+import org.hswebframework.web.id.IDGenerator;
+import org.hswebframework.web.message.MessageSubscribe;
+import org.hswebframework.web.message.Messager;
+import org.hswebframework.web.message.builder.StaticMessageBuilder;
+import org.hswebframework.web.message.builder.StaticMessageSubjectBuilder;
+import org.hswebframework.web.message.support.ObjectMessage;
+import org.hswebframework.web.service.schedule.ScheduleJobExecutor;
+import org.hswebframework.web.service.schedule.ScheduleJobService;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.lang.ref.Reference;
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.hswebframework.web.message.builder.StaticMessageSubjectBuilder.*;
+
+/**
+ * @author zhouhao
+ * @since 3.0
+ */
+@Slf4j
+public class MessagerScheduleJobExecutor implements ScheduleJobExecutor {
+    private ScheduleJobService scheduleJobService;
+
+    private ScheduleJobExecutor executor;
+
+    private Messager messager;
+
+    public MessagerScheduleJobExecutor(ScheduleJobService scheduleJobService, ScheduleJobExecutor executor, Messager messager) {
+        this.executor = executor;
+        this.messager = messager;
+        this.scheduleJobService = scheduleJobService;
+    }
+
+    public void setTags(List<String> tags) {
+        this.tags = tags;
+    }
+
+    public void setMessager(Messager messager) {
+        this.messager = messager;
+    }
+
+    public void setExecutor(ScheduleJobExecutor executor) {
+        this.executor = executor;
+    }
+
+    public void setScheduleJobService(ScheduleJobService scheduleJobService) {
+        this.scheduleJobService = scheduleJobService;
+    }
+
+    private List<String> tags = new ArrayList<>();
+
+    private List<MessageSubscribe> subscribes = new ArrayList<>();
+
+    @PreDestroy
+    public void destroy() {
+        subscribes.forEach(MessageSubscribe::cancel);
+    }
+
+    @PostConstruct
+    public void init() {
+        for (String tag : tags) {
+            MessageSubscribe subscribe = messager.subscribe(queue("quartz-job-" + tag + "-execute"))
+                    .onMessage(msg -> {
+                        JobExecutorMessage executorMessage = (JobExecutorMessage) msg;
+                        JobExecuteResultMessage resultMessage = new JobExecuteResultMessage();
+                        resultMessage.setExecuteId(executorMessage.getExecuteId());
+                        try {
+                            Object result = executor.doExecuteJob(executorMessage.getJobId(), executorMessage.getParameters());
+                            resultMessage.setResult(result);
+                        } catch (Exception e) {
+                            resultMessage.setResult(e.getMessage());
+                            resultMessage.setSuccess(false);
+                        }
+                        messager.publish(resultMessage)
+                                .to(queue("quartz-job-" + executorMessage.getExecuteId() + "-result"))
+                                .send();
+                    });
+            subscribes.add(subscribe);
+        }
+    }
+
+    @Override
+    public Object doExecuteJob(String jobId, Map<String, Object> parameter) {
+        ScheduleJobEntity jobEntity = scheduleJobService.selectByPk(jobId);
+        if (null == jobEntity) {
+            return null;
+        }
+        if (jobEntity.getTags() == null) {
+            return executor.doExecuteJob(jobId, parameter);
+        }
+        List<String> confTags = Arrays.asList(jobEntity.getTags().split("[,]"));
+        if (confTags.isEmpty() || confTags.stream().anyMatch(tags::contains)) {
+            return executor.doExecuteJob(jobId, parameter);
+        }
+        CountDownLatch latch = new CountDownLatch(1);
+        Object[] object = new Object[1];
+        Random random = new Random();
+
+        String tag = confTags.get(random.nextInt(confTags.size()));
+        String executeId = IDGenerator.MD5.generate();
+        //先订阅执行结果
+        MessageSubscribe subscribe = messager.subscribe(queue("quartz-job-" + executeId + "-result"))
+                .<JobExecuteResultMessage>onMessage(message -> {
+                    try {
+                        object[0] = ((JobExecuteResultMessage) message).getResult();
+                    } catch (Exception e) {
+                        object[0] = e.getMessage();
+                    } finally {
+                        latch.countDown();
+                    }
+                });
+        try {
+            //发送任务执行
+            messager.publish(new JobExecutorMessage(executeId, jobId, parameter))
+                    .to(queue("quartz-job-" + tag + "-execute"))
+                    .send();
+            boolean success = latch.await(1, TimeUnit.HOURS);
+            if (!success) {
+                log.warn("await job execute fail");
+            }
+        } catch (InterruptedException e) {
+            log.error(e.getMessage(), e);
+            Thread.currentThread().interrupt();
+        } finally {
+            //取消结果订阅
+            subscribe.cancel();
+        }
+        return object[0];
+    }
+}

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/pom.xml

@@ -27,19 +27,25 @@
             <artifactId>hsweb-system-schedule-service-simple</artifactId>
             <version>${project.version}</version>
         </dependency>
-      
+
         <dependency>
             <groupId>org.hswebframework.web</groupId>
             <artifactId>hsweb-system-schedule-dao-mybatis</artifactId>
             <version>${project.version}</version>
         </dependency>
-      
+
         <dependency>
             <groupId>org.hswebframework.web</groupId>
             <artifactId>hsweb-system-schedule-controller</artifactId>
             <version>${project.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.hswebframework.web</groupId>
+            <artifactId>hsweb-message-memory</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-context-support</artifactId>
@@ -50,7 +56,7 @@
             <artifactId>h2</artifactId>
             <scope>test</scope>
         </dependency>
-      
+
         <dependency>
             <groupId>com.alibaba</groupId>
             <artifactId>druid</artifactId>
@@ -78,6 +84,6 @@
             <version>2.5</version>
             <scope>test</scope>
         </dependency>
-      
+
     </dependencies>
 </project>

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/src/main/java/org/hswebframework/web/schedule/configuration/AutoCreateTable.java

@@ -7,6 +7,8 @@ import org.hswebframework.web.datasource.DataSourceHolder;
 import org.hswebframework.web.datasource.DatabaseType;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.CommandLineRunner;
+import org.springframework.core.Ordered;
+import org.springframework.core.annotation.Order;
 import org.springframework.core.io.Resource;
 import org.springframework.core.io.support.PathMatchingResourcePatternResolver;
 import org.springframework.util.StringUtils;
@@ -18,10 +20,9 @@ import java.io.Reader;
 import java.util.List;
 
 /**
- * TODO 完成注释
- *
  * @author zhouhao
  */
+@Order(Ordered.HIGHEST_PRECEDENCE)
 public class AutoCreateTable implements CommandLineRunner {
 
     @Autowired
@@ -32,7 +33,6 @@ public class AutoCreateTable implements CommandLineRunner {
         if (sqlExecutor.tableExists("QRTZ_LOCKS")) {
             return;
         }
-
         DatabaseType databaseType = DataSourceHolder.currentDatabaseType();
         String databaseTypeName = databaseType.name();
         if (databaseType == DatabaseType.jtds_sqlserver) {

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/src/main/java/org/hswebframework/web/schedule/configuration/ScheduleAutoConfiguration.java

@@ -1,6 +1,12 @@
 package org.hswebframework.web.schedule.configuration;
 
+import lombok.extern.slf4j.Slf4j;
+import org.hswebframework.web.message.Messager;
+import org.hswebframework.web.service.schedule.ScheduleJobExecutor;
+import org.hswebframework.web.service.schedule.ScheduleJobService;
+import org.hswebframework.web.service.schedule.simple.DefaultScriptScheduleJobExecutor;
 import org.hswebframework.web.service.schedule.simple.DynamicJobFactory;
+import org.hswebframework.web.service.schedule.simple.cluster.MessagerScheduleJobExecutor;
 import org.quartz.Calendar;
 import org.quartz.Scheduler;
 import org.quartz.SchedulerListener;
@@ -28,6 +34,7 @@ import java.util.Map;
 @ConditionalOnMissingBean({Scheduler.class, SchedulerFactoryBean.class})
 @ComponentScan({"org.hswebframework.web.service.schedule.simple"
         , "org.hswebframework.web.controller.schedule"})
+@Slf4j
 public class ScheduleAutoConfiguration {
     @Autowired
     private SchedulerProperties schedulerProperties;
@@ -47,6 +54,9 @@ public class ScheduleAutoConfiguration {
     @Autowired(required = false)
     private SchedulerListener[] schedulerListeners;
 
+    @Autowired(required = false)
+    private Messager messager;
+
     @Bean
     public JobFactory jobFactory() {
         return new DynamicJobFactory(new AdaptableJobFactory());
@@ -76,5 +86,23 @@ public class ScheduleAutoConfiguration {
         return schedulerFactoryBean;
     }
 
+    @Bean
+    @ConditionalOnMissingBean(ScheduleJobExecutor.class)
+    public ScheduleJobExecutor scheduleJobExecutor(ScheduleJobService scheduleJobService) {
+        ScheduleJobExecutor defaultExecutor = new DefaultScriptScheduleJobExecutor(scheduleJobService);
+        ScheduleJobExecutor scheduleJobExecutor = defaultExecutor;
+
+        if (schedulerProperties.isEnableCluster()) {
+            if (messager != null) {
+                MessagerScheduleJobExecutor executor = new MessagerScheduleJobExecutor(scheduleJobService, defaultExecutor, messager);
+                executor.setTags(schedulerProperties.getExecuteTags());
+                scheduleJobExecutor = executor;
+            } else {
+                log.warn("schedule job cluster enabled,please add messager dependency in your configuration");
+            }
+        }
+
+        return scheduleJobExecutor;
+    }
 
 }

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/src/main/java/org/hswebframework/web/schedule/configuration/SchedulerProperties.java

@@ -16,11 +16,14 @@
 
 package org.hswebframework.web.schedule.configuration;
 
+import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
 
+import java.util.List;
 import java.util.Properties;
 
 @ConfigurationProperties(prefix = "scheduler")
+@Data
 public class SchedulerProperties {
     private boolean autoStartup = true;
 
@@ -30,55 +33,11 @@ public class SchedulerProperties {
 
     private boolean waitOnShutdown = true;
 
-    private int startupDelay = 5;
+    private int startupDelay = 10;
 
     private Properties properties = new Properties();
 
-    public boolean isAutoStartup() {
-        return autoStartup;
-    }
+    private boolean enableCluster = false;
 
-    public void setAutoStartup(boolean autoStartup) {
-        this.autoStartup = autoStartup;
-    }
-
-    public boolean isOverwriteExistingJobs() {
-        return overwriteExistingJobs;
-    }
-
-    public void setOverwriteExistingJobs(boolean overwriteExistingJobs) {
-        this.overwriteExistingJobs = overwriteExistingJobs;
-    }
-
-    public String getBeanName() {
-        return beanName;
-    }
-
-    public void setBeanName(String beanName) {
-        this.beanName = beanName;
-    }
-
-    public boolean isWaitOnShutdown() {
-        return waitOnShutdown;
-    }
-
-    public void setWaitOnShutdown(boolean waitOnShutdown) {
-        this.waitOnShutdown = waitOnShutdown;
-    }
-
-    public Properties getProperties() {
-        return properties;
-    }
-
-    public void setProperties(Properties properties) {
-        this.properties = properties;
-    }
-
-    public int getStartupDelay() {
-        return startupDelay;
-    }
-
-    public void setStartupDelay(int startupDelay) {
-        this.startupDelay = startupDelay;
-    }
+    private List<String> executeTags;
 }

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/src/test/java/org/hswebframework/web/schedule/test/DynamicScheduleTests.java

@@ -16,8 +16,6 @@ import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
 /**
- * TODO 完成注释
- *
  * @author zhouhao
  */
 @Transactional(propagation = Propagation.NOT_SUPPORTED)
@@ -42,10 +40,11 @@ public class DynamicScheduleTests extends SimpleWebApplicationTests {
         entity.setStatus(DataStatus.STATUS_ENABLED);
         entity.setType("test");
         entity.setLanguage("javascript");
+        entity.setTags("core2");
         entity.setScript("" +
                 "org.hswebframework.web.schedule.test.DynamicScheduleTests.value.incrementAndGet();\n" +
                 "org.hswebframework.web.schedule.test.DynamicScheduleTests.counter.countDown();\n" +
-                "java.lang.System.out.println('job running...')");
+                "java.lang.System.out.println('script job running...')");
         entity.setQuartzConfig("{\"type\":\"cron\",\"config\":\"0/1 * * * * ?\"}");
         return entity;
     }

hsweb-system/hsweb-system-schedule/hsweb-system-schedule-starter/src/test/resources/application.yml

@@ -18,4 +18,7 @@ logging:
 mybatis:
   dynamic-datasource: false
 server:
-  port: 8080
+  port: 8080
+scheduler:
+  enable-cluster: true
+  execute-tags: core