扫描线程重构

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

 package com.byit.conf;
 
+import com.byit.factory.DaemonScanThreadRunHelper;
 import com.byit.thread.*;
+import com.byit.util.SpringUtil;
 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.ApplicationContext;
 import org.springframework.stereotype.Component;
 
+import java.util.Map;
+
 /**
  * @program: byit-myth-job->MythJobScheduler
  * @description: 扫描线程的生命周期
@@ -16,19 +21,22 @@ import org.springframework.stereotype.Component;
 @Component
 @Slf4j
 public class MythJobScheduler implements InitializingBean, DisposableBean {
+
     private final JobScheduleHelper jobScheduleHelper;
     private final LogScanHelper logScanHelper;
     private final RunRecordingScanHelper runRecordingScanHelper;
-    private final EmailScanHelper emailScanHelper;
+    //private final EmailScanHelper emailScanHelper;
     private final FlowScanHelper flowScanHelper;
+    private final ApplicationContext applicationContext;
+
 
     @Autowired
-    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper, EmailScanHelper emailScanHelper, FlowScanHelper flowScanHelper) {
+    public MythJobScheduler(JobScheduleHelper jobScheduleHelper, LogScanHelper logScanHelper, RunRecordingScanHelper runRecordingScanHelper, FlowScanHelper flowScanHelper, ApplicationContext applicationContext) {
         this.jobScheduleHelper = jobScheduleHelper;
         this.logScanHelper = logScanHelper;
         this.runRecordingScanHelper = runRecordingScanHelper;
-        this.emailScanHelper = emailScanHelper;
         this.flowScanHelper = flowScanHelper;
+        this.applicationContext = applicationContext;
     }
 
     /**
@@ -37,10 +45,10 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
      */
     @Override
     public void destroy() throws Exception {
+        DaemonScanThreadRunHelper.logoutThreadGroup();
         this.jobScheduleHelper.doStop();
         this.logScanHelper.doStop();
         this.runRecordingScanHelper.doStop();
-        this.emailScanHelper.doStop();
         this.flowScanHelper.doStop();
     }
 
@@ -50,11 +58,23 @@ public class MythJobScheduler implements InitializingBean, DisposableBean {
      */
     @Override
     public void afterPropertiesSet() throws Exception {
+        helperAutoAdd();
         //启用扫描线程
+        DaemonScanThreadRunHelper.runDaemonThreads();
         this.jobScheduleHelper.start();
         this.logScanHelper.start();
         this.runRecordingScanHelper.start();
-        this.emailScanHelper.start();
         this.flowScanHelper.start();
     }
+
+    /**
+     * 自动注册
+     */
+    public void helperAutoAdd(){
+        Map<String, BaseThreadRunHelper> beansOfType = applicationContext.getBeansOfType(BaseThreadRunHelper.class);
+        beansOfType.forEach((key, value) -> {
+            log.info("-------------自动加载帮助类,{}-------------",key);
+            DaemonScanThreadRunHelper.addThread(value);
+        });
+    }
 }

byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/DaemonScanThreadRunHelper.java

+package com.byit.factory;
+
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.SpringApplication;
+import org.springframework.context.annotation.AnnotationConfigApplicationContext;
+import org.springframework.context.support.ClassPathXmlApplicationContext;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 扫描线程运行帮助
+ * @author huangfu
+ */
+@Slf4j
+public class DaemonScanThreadRunHelper {
+    /**
+     * 控制一组守护线程是否运行
+     */
+    private volatile static boolean THREAD_GROUP_STOP = false;
+
+    /**
+     * 初始睡眠时间
+     */
+    private static final Long INIT_SLEEP_DATE = 5000L;
+    /**
+     * 关闭等待时间
+     */
+    private static final Long CLOSE_WAIT_TIME = 1000L;
+
+    /**
+     * 线程运行必须原料
+     */
+    private static final Map<String,BaseThreadRunHelper> THREAD_RUN_OPT = new HashMap<>(8);
+    /**
+     * 线程存储  用于停止线程
+     */
+    private static final Map<String,Thread> THREADS_MAP = new HashMap<>(8);
+
+
+
+    /**
+     * 添加一个线程运行机器
+     * @param threadRunHelper 线程运行实例
+     */
+    public static void addThread(BaseThreadRunHelper threadRunHelper){
+        String lockName = threadRunHelper.getLockName();
+        THREAD_RUN_OPT.put(lockName,threadRunHelper);
+    }
+
+    /**
+     * 运行线程
+     */
+    public static void runDaemonThreads(){
+        Set<Map.Entry<String, BaseThreadRunHelper>> entries = THREAD_RUN_OPT.entrySet();
+        entries.forEach(threadExamples ->{
+            String examplesKey = threadExamples.getKey();
+            BaseThreadRunHelper examplesValue = threadExamples.getValue();
+            buildThread(examplesKey,examplesValue);
+        });
+    }
+
+    /**
+     * 注销线程组
+     */
+    public static void logoutThreadGroup(){
+        DaemonScanThreadRunHelper.THREAD_GROUP_STOP = true;
+        Set<Map.Entry<String, Thread>> threadExamples = THREADS_MAP.entrySet();
+        threadExamples.forEach(threadExample ->{
+            log.warn("------------开始注销线程{}------------",threadExample);
+            String threadName = threadExample.getKey();
+            Thread thread = threadExample.getValue();
+            dateAligned(CLOSE_WAIT_TIME,threadName);
+            //判断线程是否处于终止状态
+            if(thread.getState() != Thread.State.TERMINATED){
+                thread.interrupt();
+
+                try {
+                    thread.join();
+                } catch (InterruptedException e) {
+                    e.printStackTrace( );
+                }
+            }
+            log.warn("--------{}线程被注销---------",threadName);
+        });
+    }
+
+    /**
+     * 线程构建
+     * @param lockName 行锁名称
+     * @param examplesValue 线程运行资源
+     * @return
+     */
+    private static void buildThread(String lockName, BaseThreadRunHelper examplesValue){
+        log.info("-----------开始构建扫描线程----------------");
+        DataSource dataSource = examplesValue.getDataSource();
+        if(dataSource != null){
+            String threadClassName = examplesValue.getClass().getSimpleName();
+            String threadName = "myth-job#【"+threadClassName+"】";
+            Thread exampleThread = new Thread(() ->{
+                //初始化睡眠
+                dateAligned(INIT_SLEEP_DATE,threadName);
+                log.info("---------------{}线程启动成功----------------",threadName);
+                while (!THREAD_GROUP_STOP){
+                    //定义睡眠变量
+                    boolean isSleep = false;
+                    Connection conn = null;
+                    Boolean connAutoCommit = null;
+                    PreparedStatement preparedStatement = null;
+                    try {
+                        /**
+                         * 添加行锁
+                         */
+                        conn = dataSource.getConnection();
+                        connAutoCommit = conn.getAutoCommit();
+                        conn.setAutoCommit(false);
+                        preparedStatement = conn.prepareStatement("SELECT * FROM JOB_LOCK WHERE LOCK_NAME = '"+lockName+"' FOR UPDATE ");
+                        preparedStatement.execute();
+                        //调用业务操作
+                        isSleep = examplesValue.start();
+                    }catch (Exception e){
+                        if(!THREAD_GROUP_STOP){
+                            e.printStackTrace();
+
+                        }
+                    }finally {
+                        //提交行锁
+                        if (conn != null) {
+                            try{
+                                conn.commit();
+                            }catch (Exception e){
+                                if(!THREAD_GROUP_STOP){
+                                    log.error("--------------------【提交行锁出错】---------------------");
+                                }
+                            }
+                        }
+                        //恢复自动提交
+                        if (conn != null) {
+                            try{
+                                conn.setAutoCommit(connAutoCommit);
+                            }catch (Exception e){
+                                if(!THREAD_GROUP_STOP){
+                                    log.error("--------------------【恢复自动提交出错】---------------------");
+                                }
+                            }
+                        }
+                        //关闭数据库执行器
+                        if(preparedStatement != null){
+                            try {
+                                preparedStatement.close();
+                            } catch (SQLException e) {
+                                if(!THREAD_GROUP_STOP){
+                                    log.error("--------------------【关闭执行器出错】---------------------");
+                                }
+                            }
+                        }
+                        //关闭数据库连接
+                        if(conn != null){
+                            try {
+                                conn.close();
+                            } catch (SQLException e) {
+                                if(!THREAD_GROUP_STOP){
+                                    log.error("--------------------【关闭执行器出错】---------------------");
+                                }
+                            }
+                        }
+                    }
+
+                    if(isSleep){
+                        dateAligned(examplesValue.getSleepTime(),threadName);
+                    }
+                }
+            });
+
+            exampleThread.setName(threadName);
+            exampleThread.setDaemon(true);
+            exampleThread.start();
+            THREADS_MAP.put(threadName,exampleThread);
+        }else{
+            System.out.println("--------------出错了-----------");
+        }
+    }
+
+    /**
+     * 对齐时钟。整秒运行
+     * @param waitTime 休眠时间
+     * @param threadName 线程名称
+     */
+    private static void dateAligned(long waitTime,String threadName){
+        try {
+            log.debug("-----------线程{}开始休眠，休眠时间{}---------",threadName,waitTime);
+            TimeUnit.MILLISECONDS.sleep(waitTime - System.currentTimeMillis()%1000);
+        } catch (InterruptedException e) {
+            log.warn("----------------【{}线程被中断】-----------------------",threadName);
+        }
+    }
+
+}
+

byit-myth-core/myth-admin-core/src/main/java/com/byit/factory/material/ThreadRunMaterial.java

+package com.byit.factory.material;
+
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.ToString;
+
+import javax.sql.DataSource;
+
+/**
+ * @author huangfu
+ */
+@AllArgsConstructor
+@Data
+@ToString
+public class ThreadRunMaterial {
+    /**
+     * 此线程
+     */
+    private Thread thread;
+    /**
+     * 业务规则
+     */
+    private BaseThreadRunHelper threadRunHelper;
+    /**
+     * 循环一轮睡眠时长
+     */
+    private Long sleepTime;
+
+}

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

+package com.byit.thread;
+
+import javax.sql.DataSource;
+
+/**
+ * @author 线程运行规范
+ */
+public abstract class BaseThreadRunHelper {
+    /**
+     * 具体要做的业务操作
+     * @return 返回是否需要睡眠
+     */
+    public abstract boolean start();
+
+    /**
+     * 返回数据库运行实例
+     * @return 数据路连接
+     */
+    public abstract DataSource getDataSource();
+
+    /**
+     * 返回行锁名称
+     * @return 返回行锁名称
+     */
+    public abstract String getLockName();
+
+    /**
+     * 返回循环一轮的睡眠时间
+     * @return 睡眠时长
+     */
+    public long getSleepTime(){
+        return 20000;
+    }
+}

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

+package com.byit.thread.helper;
+
+import cn.hutool.core.collection.CollectionUtil;
+import com.byit.factory.DaemonScanThreadRunHelper;
+import com.byit.model.EmailAlarm;
+import com.byit.model.vo.EmailAlarmVo;
+import com.byit.service.EmailAlarmService;
+import com.byit.thread.BaseThreadRunHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.BeanUtils;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+import java.util.List;
+
+/**
+ * @author huangfu
+ */
+@Component
+@Slf4j
+public class EmailThreadRunHelper extends BaseThreadRunHelper {
+    private static final String LOCK_NAME = "emali_alarm_lock";
+    private final DataSource dataSource;
+    private final EmailAlarmService emailAlarmService;
+
+    public EmailThreadRunHelper(DataSource dataSource, EmailAlarmService emailAlarmService) {
+        this.dataSource = dataSource;
+        this.emailAlarmService = emailAlarmService;
+    }
+
+    @Override
+    public boolean start() {
+        boolean isSleep = false;
+        //查询未告警的邮箱
+        List<EmailAlarm> emailAlarmByAlarmResults = emailAlarmService.findEmailAlarmByAlarmResult();
+        if(CollectionUtil.isNotEmpty(emailAlarmByAlarmResults)){
+            emailAlarmByAlarmResults.forEach(emailAlarm ->{
+                EmailAlarmVo emailAlarmVo = new EmailAlarmVo();
+                BeanUtils.copyProperties(emailAlarm,emailAlarmVo);
+                emailAlarmService.sendEmail(emailAlarmVo);
+            });
+        }else{
+            isSleep = true;
+        }
+        return isSleep;
+    }
+
+    @Override
+    public DataSource getDataSource() {
+        return this.dataSource;
+    }
+
+    @Override
+    public String getLockName() {
+        return LOCK_NAME;
+    }
+}

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

 package com.byit.util;
 
+import com.byit.thread.BaseThreadRunHelper;
 import org.springframework.beans.BeansException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.stereotype.Component;
 
+import java.util.Map;
+
 /**
  * Spring工具类
  * @author huangfu
@@ -58,4 +61,14 @@ public class SpringUtil implements ApplicationContextAware {
     public static <T> T getBean(String name,Class<T> clazz){
         return getApplicationContext().getBean(name, clazz);
     }
+
+    /**
+     * 获取实现某个接口的类
+     * @param clazz
+     * @param <T>
+     * @return
+     */
+    public static <T>  Map<String,T> getBeansOfType(Class<T> clazz){
+        return getApplicationContext().getBeansOfType(clazz);
+    }
 }
\ No newline at end of file