线程循环访问数据库查询即将执行的数据库（还有BUG，任务节点被重复读取，考虑增加任务重复次数来解决这个问题）

byit-myth-admin/pom.xml

@@ -30,6 +30,7 @@
             <groupId>com.alibaba</groupId>
             <artifactId>fastjson</artifactId>
         </dependency>
+
     </dependencies>
 
 </project>
\ No newline at end of file

byit-myth-admin/src/main/java/com/byit/conf/DataSourceConf.java

+package com.byit.conf;
+
+import com.byit.service.JobInfoService;
+import lombok.Data;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import javax.sql.DataSource;
+
+/**
+ * @program: byit-myth-job->DataSourceConf
+ * @description: TODO
+ * @author: huangfu
+ * @date: 2019/12/9 19:52
+ **/
+@Data
+@Component("dataSourceConf")
+public class DataSourceConf  {
+
+    @Resource
+    private DataSource dataSource;
+}

byit-myth-admin/src/main/java/com/byit/conf/MythJobScheduler.java

+package com.byit.conf;
+
+import com.byit.thread.JobScheduleHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+import javax.sql.DataSource;
+
+/**
+ * @program: byit-myth-job->MythJobScheduler
+ * @description: 扫描线程的生命周期
+ * @author: huangfu
+ * @date: 2019/12/9 16:53
+ **/
+@Component
+@Slf4j
+@DependsOn("dataSourceConf")
+public class MythJobScheduler implements InitializingBean, DisposableBean {
+    @Autowired
+    DataSourceConf dataSourceConf;
+    @Autowired
+    private JobScheduleHelper jobScheduleHelper;
+    /**
+     * 销毁方法
+     * @throws Exception
+     */
+    @Override
+    public void destroy() throws Exception {
+
+    }
+
+    /**
+     * 初始化方法
+     * @throws Exception
+     */
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        jobScheduleHelper.setDataSource(dataSourceConf.getDataSource());
+        //启用扫描线程
+        jobScheduleHelper.start();
+    }
+}

byit-myth-admin/src/main/java/com/byit/controller/JobController.java

@@ -8,6 +8,7 @@ import com.byit.service.JobInfoService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 
+import javax.sql.DataSource;
 import java.util.List;
 
 /**
@@ -20,6 +21,8 @@ import java.util.List;
 @RequestMapping("job")
 public class JobController {
     @Autowired
+    private DataSource dataSource;
+    @Autowired
     private JobInfoService jobInfoService;
     @PostMapping(value = "addJob")
     public String addJob(@RequestBody PluginBeanJobInfo pluginBeanJobInfo){

byit-myth-admin/src/main/resources/application.yml

+spring:
+  datasource:
+    driver-class-name: com.mysql.jdbc.Driver
+    url: jdbc:mysql://10.0.10.118:3306/myth-job?Unicode=true&characterEncoding=UTF-8&useSSL=true
+    username: root
+    password: 123456
+
+mybatis:
+  mapper-locations: /mapper/*.xml
\ No newline at end of file

byit-myth-core/myth-admin-core/pom.xml

@@ -56,6 +56,13 @@
             <artifactId>spring-boot-configuration-processor</artifactId>
         </dependency>
 
+        <!-- https://mvnrepository.com/artifact/org.springframework.boot/spring-boot-starter-jdbc -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-jdbc</artifactId>
+        </dependency>
+
+
 
     </dependencies>
 

byit-myth-core/myth-admin-core/src/main/java/com/byit/conf/MythJobAutoConfigure.java

@@ -3,9 +3,19 @@ package com.byit.conf;
 import com.byit.service.impl.JobInfoServiceImpl;
 import com.byit.service.JobInfoService;
 import org.mybatis.spring.annotation.MapperScan;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-import org.springframework.context.annotation.PropertySource;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.AutoConfigureBefore;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceBuilder;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.*;
+import org.springframework.jdbc.datasource.DriverManagerDataSource;
+
+import javax.sql.DataSource;
 
 /**
  * @program: byit-myth-job->AppConf
@@ -16,8 +26,4 @@ import org.springframework.context.annotation.PropertySource;
 @Configuration
 @MapperScan("com.byit.dao")
 public class MythJobAutoConfigure {
-    @Bean
-    public JobInfoService jobInfoService(){
-        return new JobInfoServiceImpl();
-    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/service/impl/JobInfoServiceImpl.java

@@ -3,6 +3,7 @@ package com.byit.service.impl;
 import com.byit.dao.JobInfoMapper;
 import com.byit.job.model.MythJobInfo;
 import com.byit.service.JobInfoService;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
@@ -14,6 +15,8 @@ import java.util.List;
  * @author: huangfu
  * @date: 2019/12/9 15:18
  **/
+@Service
+@Slf4j
 public class JobInfoServiceImpl implements JobInfoService {
 
     @Autowired

byit-myth-core/myth-admin-core/src/main/java/com/byit/thread/JobScheduleHelper.java

+package com.byit.thread;
+
+import cn.hutool.core.collection.CollectionUtil;
+import cn.hutool.core.util.ObjectUtil;
+import com.byit.job.model.MythJobInfo;
+import com.byit.service.JobInfoService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @program: byit-myth-job->JobScheduleHeloer
+ * @description: 工作时间排期表，这里开启了两条线程：
+ *              一条线程去任务节点读取七秒内将要执行的任务添加进任务排期表
+ *              一条线程去任务排期表中预读五秒将要执行的任务添加进任务调度轮盘
+ * @author: huangfu
+ * @date: 2019/12/9 11:16
+ **/
+@Slf4j
+@Component
+public class JobScheduleHelper{
+    private DataSource dataSource;
+    @Autowired
+    private JobInfoService jobInfoService;
+
+    /**
+     * 读取任务节点的预读
+     */
+    public static final long PRE_READ_MS = 7000;
+    /**
+     * 读取任务排期表的预读
+     */
+    public static final long SCHEDULE_READ_MS=5000;
+    /**
+     * 任务节点线程
+     */
+    private Thread jobInfoThread;
+    /**
+     * 任务排期表线程
+     */
+    private Thread scheduleThread;
+
+    /**
+     * 是否停止扫描任务节点表
+     */
+    private volatile boolean jobInfoThreadToStop = false;
+    /**
+     * 是否停止排期表的扫描，停止向任务调度轮盘添加任务
+     */
+    private volatile boolean scheduleThreadToStop = false;
+
+    /**
+     * 启动两条线程
+     */
+    public void start(){
+        jobInfoThread = new Thread(()->{
+            try {
+                TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
+            } catch (InterruptedException e) {
+                if (!jobInfoThreadToStop) {
+                    log.error(e.getMessage(), e);
+                }
+            }
+            log.info("---------------------init myth-job admin jobInfoThread success------------------------");
+
+            while (!jobInfoThreadToStop) {
+                //开始去扫描任务节点
+                long start = System.currentTimeMillis();
+                Connection conn = null;
+                Boolean connAutoCommit = null;
+                PreparedStatement preparedStatement = null;
+
+                boolean preReadSuc = true;
+
+                try{
+                    conn = dataSource.getConnection();
+                    //记录当前的自动提交状态
+                    connAutoCommit = conn.getAutoCommit();
+                    //修改为不自动提交
+                    conn.setAutoCommit(false);
+                    //先添加扫描任务节点的行锁
+                    preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = 'scanning_job_nodes' FOR UPDATE ");
+                    preparedStatement.execute();
+                    //行锁已经加上  后续处理
+                    long nowTime = System.currentTimeMillis();
+                    //开始寻找此时 不是暂停状态，而且七秒内即将运行的任务
+                    List<MythJobInfo> mythJobInfoByTriggerNextTime = jobInfoService.findMythJobInfoByTriggerNextTime(nowTime + PRE_READ_MS);
+                    log.info("----------------{}----------------",mythJobInfoByTriggerNextTime);
+                    if(CollectionUtil.isNotEmpty(mythJobInfoByTriggerNextTime)){
+                        mythJobInfoByTriggerNextTime.forEach(e -> System.out.println(e ));
+                    }else{
+                        preReadSuc = false;
+                    }
+                }catch (Exception e){
+                    if(!jobInfoThreadToStop){
+                        e.printStackTrace();
+                        log.error("----------------------{}----------------------",e.getMessage());
+                    }
+                }finally {
+                    //commit
+                    if (conn!=null) {
+                        try {
+                            conn.commit();
+                        }catch (SQLException e){
+                            if(!jobInfoThreadToStop){
+                                log.error("----------------------{}----------------------",e.getMessage());
+                            }
+                        }
+                        //设置提交状态恢复原来的值
+                        try {
+                            conn.setAutoCommit(connAutoCommit);
+                        }catch (SQLException e){
+                            if(!jobInfoThreadToStop){
+                                log.error("----------------------{}----------------------",e.getMessage());
+                            }
+                        }
+                        //关闭连接
+                        try {
+                            conn.close();
+                        }catch (SQLException e){
+                            if(!jobInfoThreadToStop){
+                                log.error("----------------------{}----------------------",e.getMessage());
+                            }
+                        }
+                    }
+
+                    //关闭执行器
+                    if(null !=preparedStatement){
+                        try {
+                            preparedStatement.close();
+                        } catch (SQLException e) {
+                            if(!jobInfoThreadToStop){
+                                log.error("----------------------{}----------------------",e.getMessage());
+                            }
+                        }
+                    }
+                }
+                //计算总的处理时间  将时间保持在一秒处理一次
+                long cost = System.currentTimeMillis()-start;
+                //不足一秒的等待 等够1秒为止
+                if(cost < 1000){
+                    // 预读期：成功-每秒扫描一次；失败-跳过这段时间
+                    try {
+                        TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
+                    } catch (InterruptedException e) {
+                        if (!jobInfoThreadToStop) {
+                            log.error("----------------------{}-------------------------",e.getMessage());
+                        }
+                    }
+                }
+            }
+        });
+        //设置为守护线程
+        jobInfoThread.setDaemon(true);
+        //设置名字
+        jobInfoThread.setName("myth-job,admin JobScheduleHelper#scheduleThread");
+        jobInfoThread.start();
+    }
+
+    public void setDataSource(DataSource dataSource) {
+        this.dataSource = dataSource;
+    }
+
+    public static void main(String[] args) {
+        System.out.println(System.currentTimeMillis()+7000 );
+    }
+}

byit-myth-core/myth-admin-core/src/main/java/com/byit/util/SpringUtil.java

+package com.byit.util;
+
+import org.springframework.beans.BeansException;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.ApplicationContextAware;
+import org.springframework.stereotype.Component;
+
+/**
+ * Spring工具类
+ * @author huangfu
+ */
+@Component
+public class SpringUtil implements ApplicationContextAware {
+
+    private static ApplicationContext applicationContext = null;
+
+    @Override
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
+        if(SpringUtil.applicationContext == null) {
+            SpringUtil.applicationContext = applicationContext;
+        }
+    }
+
+    /**
+     * 获取applicationContext
+     * @return 返回applicationContext
+     */
+    public static ApplicationContext getApplicationContext() {
+        return applicationContext;
+    }
+
+    /**
+     * 通过name获取 Bean.
+     * @param name
+     * @return
+     */
+    public static Object getBean(String name){
+        return getApplicationContext().getBean(name);
+    }
+
+    /**
+     * 通过class获取Bean.
+     * @param clazz
+     * @param <T>
+     * @return
+     */
+    public static <T> T getBean(Class<T> clazz){
+        return getApplicationContext().getBean(clazz);
+    }
+
+    /**
+     * 通过name,以及Clazz返回指定的Bean
+     * @param name
+     * @param clazz
+     * @param <T>
+     * @return
+     */
+    public static <T> T getBean(String name,Class<T> clazz){
+        return getApplicationContext().getBean(name, clazz);
+    }
+}
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/resources/META-INF/spring.factories

-org.springframework.boot.autoconfigure.EnableAutoConfiguration=\com.byit.conf.MythJobAutoConfigure
\ No newline at end of file

byit-myth-core/myth-admin-core/src/main/resources/mapper/MythJobInfoMapper.xml

@@ -30,7 +30,8 @@
         t.author,
         t.add_time,
         t.update_time,
-        t.removal_mark
+        t.removal_mark,
+        t.repeat_times
     </sql>
 
     <resultMap id="MythJobInfo" type="com.byit.job.model.MythJobInfo" >
@@ -63,6 +64,7 @@
         <result column="add_time" property="addTime"/>
         <result column="update_time" property="updateTime"/>
         <result column="removal_mark" property="removalMark"/>
+        <result column="repeat_times" property="repeatTimes"/>
     </resultMap>
 
     <select id="findMythJobInfoByTriggerNextTime" resultMap="MythJobInfo">

byit-myth-core/myth-core-common/src/main/java/com/byit/job/model/MythJobInfo.java

@@ -141,4 +141,8 @@ public class MythJobInfo {
      * 删除标志
      */
     private String removalMark;
+    /**
+     * 重复次数   -1遵循cron表达式解析
+     */
+    private int repeatTimes;
 }

pom.xml

@@ -162,8 +162,6 @@
             </dependency>
 
 
-
-
         </dependencies>
     </dependencyManagement>