review job helper

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobFailMonitorHelper.java

@@ -49,79 +49,78 @@ public class JobFailMonitorHelper {
      */
     @PostConstruct
     public void start() {
-        monitorThread = new Thread(new Runnable() {
-
-            @Override
-            public void run() {
-
-                // monitor
-                while (!toStop) {
-                    try {
-
-                        List<Long> failLogIds = xxlJobLogDao.findFailJobLogIds(1000);
-                        if (failLogIds != null && !failLogIds.isEmpty()) {
-                            for (long failLogId : failLogIds) {
-
-                                // lock log
-                                int lockRet = xxlJobLogDao.updateAlarmStatus(failLogId, 0, -1);
-                                if (lockRet < 1) {
-                                    continue;
-                                }
-                                XxlJobLog log = xxlJobLogDao.load(failLogId);
-                                XxlJobInfo info = xxlJobInfoDao.loadById(log.getJobId());
-
-                                // 1、fail retry monitor
-                                if (log.getExecutorFailRetryCount() > 0) {
-                                    jobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount() - 1), log.getExecutorShardingParam(), log.getExecutorParam());
-                                    String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>" + I18nUtil.getString("jobconf_trigger_type_retry") + "<<<<<<<<<<< </span><br>";
-                                    log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
-                                    xxlJobLogDao.updateTriggerInfo(log);
-                                }
-
-                                // 2、fail alarm monitor
-                                int newAlarmStatus = 0;        // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败
-                                if (info != null && info.getAlarmEmail() != null && info.getAlarmEmail().trim().length() > 0) {
-                                    boolean alarmResult = true;
-                                    try {
-                                        alarmResult = failAlarm(info, log);
-                                    } catch (Exception e) {
-                                        alarmResult = false;
-                                        logger.error(e.getMessage(), e);
-                                    }
-                                    newAlarmStatus = alarmResult ? 2 : 3;
-                                } else {
-                                    newAlarmStatus = 1;
-                                }
-
-                                xxlJobLogDao.updateAlarmStatus(failLogId, -1, newAlarmStatus);
-                            }
-                        }
+        monitorThread = new Thread(this::failJobMonitor);
+        monitorThread.setDaemon(true);
+        monitorThread.setName("jobAdminFailMonitorHp");
+        //monitorThread.setName("xxl-job, admin JobFailMonitorHelper");
+        monitorThread.start();
+    }
+
 
-                    } catch (Exception e) {
-                        if (!toStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
+    private void failJobMonitor() {
+        // monitor
+        while (!toStop) {
+            try {
+                List<Long> failLogIds = xxlJobLogDao.findFailJobLogIds(1000);
+                if (failLogIds != null && !failLogIds.isEmpty()) {
+                    for (long failLogId : failLogIds) {
+
+                        // lock log
+                        int lockRet = xxlJobLogDao.updateAlarmStatus(failLogId, 0, -1);
+                        if (lockRet < 1) {
+                            continue;
+                        }
+                        XxlJobLog log = xxlJobLogDao.load(failLogId);
+                        XxlJobInfo info = xxlJobInfoDao.loadById(log.getJobId());
+
+                        // 1、fail retry monitor
+                        if (log.getExecutorFailRetryCount() > 0) {
+                            jobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount() - 1), log.getExecutorShardingParam(), log.getExecutorParam());
+                            String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>" + I18nUtil.getString("jobconf_trigger_type_retry") + "<<<<<<<<<<< </span><br>";
+                            log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
+                            xxlJobLogDao.updateTriggerInfo(log);
                         }
-                    }
 
-                    try {
-                        TimeUnit.SECONDS.sleep(10);
-                    } catch (Exception e) {
-                        if (!toStop) {
-                            logger.error(e.getMessage(), e);
+                        // 2、fail alarm monitor
+                        int newAlarmStatus = 0;        // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败
+                        if (info != null && info.getAlarmEmail() != null && info.getAlarmEmail().trim().length() > 0) {
+                            boolean alarmResult = true;
+                            try {
+                                alarmResult = failAlarm(info, log);
+                            } catch (Exception e) {
+                                alarmResult = false;
+                                logger.error(e.getMessage(), e);
+                            }
+                            newAlarmStatus = alarmResult ? 2 : 3;
+                        } else {
+                            newAlarmStatus = 1;
                         }
-                    }
 
+                        xxlJobLogDao.updateAlarmStatus(failLogId, -1, newAlarmStatus);
+                    }
                 }
 
-                logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop");
+            } catch (Exception e) {
+                if (!toStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
+                }
+            }
 
+            try {
+                TimeUnit.SECONDS.sleep(10);
+            } catch (Exception e) {
+                if (!toStop) {
+                    logger.error(e.getMessage(), e);
+                }
             }
-        });
-        monitorThread.setDaemon(true);
-        monitorThread.setName("xxl-job, admin JobFailMonitorHelper");
-        monitorThread.start();
+
+        }
+
+        logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop");
+
     }
 
+
     @PreDestroy
     public void toStop() {
         toStop = true;

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobLogReportHelper.java

@@ -46,112 +46,115 @@ public class JobLogReportHelper {
     @PostConstruct
     public void start() {
         logrThread = new Thread(new Runnable() {
-
             @Override
             public void run() {
-
-                // last clean log time
-                long lastCleanLogTime = 0;
+                jobLogReport();
+            }
+        });
+        logrThread.setDaemon(true);
+        logrThread.setName("xxl-job, admin JobLogReportHelper");
+        logrThread.start();
+    }
 
 
-                while (!toStop) {
+    private void jobLogReport() {
+        // last clean log time
+        long lastCleanLogTime = 0;
 
-                    // 1、log-report refresh: refresh log report in 3 days
-                    try {
 
-                        for (int i = 0; i < 3; i++) {
+        while (!toStop) {
 
-                            // today
-                            Calendar itemDay = Calendar.getInstance();
-                            itemDay.add(Calendar.DAY_OF_MONTH, -i);
-                            itemDay.set(Calendar.HOUR_OF_DAY, 0);
-                            itemDay.set(Calendar.MINUTE, 0);
-                            itemDay.set(Calendar.SECOND, 0);
-                            itemDay.set(Calendar.MILLISECOND, 0);
+            // 1、log-report refresh: refresh log report in 3 days
+            try {
 
-                            Date todayFrom = itemDay.getTime();
+                for (int i = 0; i < 3; i++) {
 
-                            itemDay.set(Calendar.HOUR_OF_DAY, 23);
-                            itemDay.set(Calendar.MINUTE, 59);
-                            itemDay.set(Calendar.SECOND, 59);
-                            itemDay.set(Calendar.MILLISECOND, 999);
+                    // today
+                    Calendar itemDay = Calendar.getInstance();
+                    itemDay.add(Calendar.DAY_OF_MONTH, -i);
+                    itemDay.set(Calendar.HOUR_OF_DAY, 0);
+                    itemDay.set(Calendar.MINUTE, 0);
+                    itemDay.set(Calendar.SECOND, 0);
+                    itemDay.set(Calendar.MILLISECOND, 0);
 
-                            Date todayTo = itemDay.getTime();
+                    Date todayFrom = itemDay.getTime();
 
-                            // refresh log-report every minute
-                            XxlJobLogReport xxlJobLogReport = new XxlJobLogReport();
-                            xxlJobLogReport.setTriggerDay(todayFrom);
-                            xxlJobLogReport.setRunningCount(0);
-                            xxlJobLogReport.setSucCount(0);
-                            xxlJobLogReport.setFailCount(0);
+                    itemDay.set(Calendar.HOUR_OF_DAY, 23);
+                    itemDay.set(Calendar.MINUTE, 59);
+                    itemDay.set(Calendar.SECOND, 59);
+                    itemDay.set(Calendar.MILLISECOND, 999);
 
-                            Map<String, Object> triggerCountMap = xxlJobLogDao.findLogReport(todayFrom, todayTo);
-                            if (triggerCountMap != null && triggerCountMap.size() > 0) {
-                                int triggerDayCount = triggerCountMap.containsKey("triggerDayCount") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCount"))) : 0;
-                                int triggerDayCountRunning = triggerCountMap.containsKey("triggerDayCountRunning") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCountRunning"))) : 0;
-                                int triggerDayCountSuc = triggerCountMap.containsKey("triggerDayCountSuc") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCountSuc"))) : 0;
-                                int triggerDayCountFail = triggerDayCount - triggerDayCountRunning - triggerDayCountSuc;
+                    Date todayTo = itemDay.getTime();
 
-                                xxlJobLogReport.setRunningCount(triggerDayCountRunning);
-                                xxlJobLogReport.setSucCount(triggerDayCountSuc);
-                                xxlJobLogReport.setFailCount(triggerDayCountFail);
-                            }
+                    // refresh log-report every minute
+                    XxlJobLogReport xxlJobLogReport = new XxlJobLogReport();
+                    xxlJobLogReport.setTriggerDay(todayFrom);
+                    xxlJobLogReport.setRunningCount(0);
+                    xxlJobLogReport.setSucCount(0);
+                    xxlJobLogReport.setFailCount(0);
 
-                            // do refresh
-                            int ret = xxlJobLogReportDao.update(xxlJobLogReport);
-                            if (ret < 1) {
-                                xxlJobLogReportDao.save(xxlJobLogReport);
-                            }
-                        }
+                    Map<String, Object> triggerCountMap = xxlJobLogDao.findLogReport(todayFrom, todayTo);
+                    if (triggerCountMap != null && triggerCountMap.size() > 0) {
+                        int triggerDayCount = triggerCountMap.containsKey("triggerDayCount") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCount"))) : 0;
+                        int triggerDayCountRunning = triggerCountMap.containsKey("triggerDayCountRunning") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCountRunning"))) : 0;
+                        int triggerDayCountSuc = triggerCountMap.containsKey("triggerDayCountSuc") ? Integer.valueOf(String.valueOf(triggerCountMap.get("triggerDayCountSuc"))) : 0;
+                        int triggerDayCountFail = triggerDayCount - triggerDayCountRunning - triggerDayCountSuc;
 
-                    } catch (Exception e) {
-                        if (!toStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, job log report thread error:{}", e);
-                        }
+                        xxlJobLogReport.setRunningCount(triggerDayCountRunning);
+                        xxlJobLogReport.setSucCount(triggerDayCountSuc);
+                        xxlJobLogReport.setFailCount(triggerDayCountFail);
                     }
 
-                    // 2、log-clean: switch open & once each day
-                    if (xxlJobAdminConfig.getLogRetentionDays() > 0 && System.currentTimeMillis() - lastCleanLogTime > 24 * 60 * 60 * 1000) {
-
-                        // expire-time
-                        Calendar expiredDay = Calendar.getInstance();
-                        expiredDay.add(Calendar.DAY_OF_MONTH, -1 * xxlJobAdminConfig.getLogRetentionDays());
-                        expiredDay.set(Calendar.HOUR_OF_DAY, 0);
-                        expiredDay.set(Calendar.MINUTE, 0);
-                        expiredDay.set(Calendar.SECOND, 0);
-                        expiredDay.set(Calendar.MILLISECOND, 0);
-                        Date clearBeforeTime = expiredDay.getTime();
-
-                        // clean expired log
-                        List<Long> logIds = null;
-                        do {
-                            logIds = xxlJobLogDao.findClearLogIds(0, 0, clearBeforeTime, 0, 1000);
-                            if (logIds != null && logIds.size() > 0) {
-                                xxlJobLogDao.clearLog(logIds);
-                            }
-                        } while (logIds != null && logIds.size() > 0);
-
-                        // update clean time
-                        lastCleanLogTime = System.currentTimeMillis();
+                    // do refresh
+                    int ret = xxlJobLogReportDao.update(xxlJobLogReport);
+                    if (ret < 1) {
+                        xxlJobLogReportDao.save(xxlJobLogReport);
                     }
+                }
+
+            } catch (Exception e) {
+                if (!toStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, job log report thread error:{}", e.getMessage());
+                }
+            }
 
-                    try {
-                        TimeUnit.MINUTES.sleep(1);
-                    } catch (Exception e) {
-                        if (!toStop) {
-                            logger.error(e.getMessage(), e);
-                        }
+            // 2、log-clean: switch open & once each day
+            if (xxlJobAdminConfig.getLogRetentionDays() > 0 && System.currentTimeMillis() - lastCleanLogTime > 24 * 60 * 60 * 1000) {
+
+                // expire-time
+                Calendar expiredDay = Calendar.getInstance();
+                expiredDay.add(Calendar.DAY_OF_MONTH, -1 * xxlJobAdminConfig.getLogRetentionDays());
+                expiredDay.set(Calendar.HOUR_OF_DAY, 0);
+                expiredDay.set(Calendar.MINUTE, 0);
+                expiredDay.set(Calendar.SECOND, 0);
+                expiredDay.set(Calendar.MILLISECOND, 0);
+                Date clearBeforeTime = expiredDay.getTime();
+
+                // clean expired log
+                List<Long> logIds = null;
+                do {
+                    logIds = xxlJobLogDao.findClearLogIds(0, 0, clearBeforeTime, 0, 1000);
+                    if (logIds != null && logIds.size() > 0) {
+                        xxlJobLogDao.clearLog(logIds);
                     }
+                } while (logIds != null && logIds.size() > 0);
+
+                // update clean time
+                lastCleanLogTime = System.currentTimeMillis();
+            }
 
+            try {
+                TimeUnit.MINUTES.sleep(1);
+            } catch (Exception e) {
+                if (!toStop) {
+                    logger.error(e.getMessage(), e);
                 }
+            }
 
-                logger.info(">>>>>>>>>>> xxl-job, job log report thread stop");
+        }
+
+        logger.info(">>>>>>>>>>> xxl-job, job log report thread stop");
 
-            }
-        });
-        logrThread.setDaemon(true);
-        logrThread.setName("xxl-job, admin JobLogReportHelper");
-        logrThread.start();
     }
 
     @PreDestroy

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryMonitorHelper.java

@@ -45,73 +45,77 @@ public class JobRegistryMonitorHelper {
         registryThread = new Thread(new Runnable() {
             @Override
             public void run() {
-                while (!toStop) {
-                    try {
-                        // auto registry group
-                        List<XxlJobGroup> groupList = xxlJobGroupDao.findByAddressType(0);
-                        if (groupList != null && !groupList.isEmpty()) {
-
-                            // remove dead address (admin/executor)
-                            List<Integer> ids = xxlJobRegistryDao.findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
-                            if (ids != null && ids.size() > 0) {
-                                xxlJobRegistryDao.removeDead(ids);
-                            }
+                jobRegistryMonitor();
+            }
+        });
+        registryThread.setDaemon(true);
+        registryThread.setName("xxl-job, admin JobRegistryMonitorHelper");
+        registryThread.start();
+    }
+
+    private void jobRegistryMonitor() {
+        while (!toStop) {
+            try {
+                // auto registry group
+                List<XxlJobGroup> groupList = xxlJobGroupDao.findByAddressType(0);
+                if (groupList != null && !groupList.isEmpty()) {
+
+                    // remove dead address (admin/executor)
+                    List<Integer> ids = xxlJobRegistryDao.findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
+                    if (ids != null && ids.size() > 0) {
+                        xxlJobRegistryDao.removeDead(ids);
+                    }
 
-                            // fresh online address (admin/executor)
-                            HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
-                            List<XxlJobRegistry> list = xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
-                            if (list != null) {
-                                for (XxlJobRegistry item : list) {
-                                    if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
-                                        String appName = item.getRegistryKey();
-                                        List<String> registryList = appAddressMap.get(appName);
-                                        if (registryList == null) {
-                                            registryList = new ArrayList<String>();
-                                        }
-
-                                        if (!registryList.contains(item.getRegistryValue())) {
-                                            registryList.add(item.getRegistryValue());
-                                        }
-                                        appAddressMap.put(appName, registryList);
-                                    }
+                    // fresh online address (admin/executor)
+                    HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
+                    List<XxlJobRegistry> list = xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
+                    if (list != null) {
+                        for (XxlJobRegistry item : list) {
+                            if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
+                                String appName = item.getRegistryKey();
+                                List<String> registryList = appAddressMap.get(appName);
+                                if (registryList == null) {
+                                    registryList = new ArrayList<String>();
                                 }
-                            }
 
-                            // fresh group address
-                            for (XxlJobGroup group : groupList) {
-                                List<String> registryList = appAddressMap.get(group.getAppName());
-                                String addressListStr = null;
-                                if (registryList != null && !registryList.isEmpty()) {
-                                    Collections.sort(registryList);
-                                    addressListStr = "";
-                                    for (String item : registryList) {
-                                        addressListStr += item + ",";
-                                    }
-                                    addressListStr = addressListStr.substring(0, addressListStr.length() - 1);
+                                if (!registryList.contains(item.getRegistryValue())) {
+                                    registryList.add(item.getRegistryValue());
                                 }
-                                group.setAddressList(addressListStr);
-                                xxlJobGroupDao.update(group);
+                                appAddressMap.put(appName, registryList);
                             }
                         }
-                    } catch (Exception e) {
-                        if (!toStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
-                        }
                     }
-                    try {
-                        TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
-                    } catch (InterruptedException e) {
-                        if (!toStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
+
+                    // fresh group address
+                    for (XxlJobGroup group : groupList) {
+                        List<String> registryList = appAddressMap.get(group.getAppName());
+                        String addressListStr = null;
+                        if (registryList != null && !registryList.isEmpty()) {
+                            Collections.sort(registryList);
+                            addressListStr = "";
+                            for (String item : registryList) {
+                                addressListStr += item + ",";
+                            }
+                            addressListStr = addressListStr.substring(0, addressListStr.length() - 1);
                         }
+                        group.setAddressList(addressListStr);
+                        xxlJobGroupDao.update(group);
                     }
                 }
-                logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
+            } catch (Exception e) {
+                if (!toStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e.getMessage());
+                }
             }
-        });
-        registryThread.setDaemon(true);
-        registryThread.setName("xxl-job, admin JobRegistryMonitorHelper");
-        registryThread.start();
+            try {
+                TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
+            } catch (InterruptedException e) {
+                if (!toStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e.getMessage());
+                }
+            }
+        }
+        logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
     }
 
 

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobScheduleHelper.java

@@ -59,233 +59,240 @@ public class JobScheduleHelper {
         scheduleThread = new Thread(new Runnable() {
             @Override
             public void run() {
+                jobScheduleStop();
+            }
+        });
+        scheduleThread.setDaemon(true);
+        scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
+        scheduleThread.start();
 
-                try {
-                    TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
-                } catch (InterruptedException e) {
-                    if (!scheduleThreadToStop) {
-                        logger.error(e.getMessage(), e);
-                    }
-                }
-                logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
-
-                // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
-                int preReadCount = (xxlJobAdminConfig.getTriggerPoolFastMax() + xxlJobAdminConfig.getTriggerPoolSlowMax()) * 20;
+        // ring thread
+        ringThread = new Thread(new Runnable() {
+            @Override
+            public void run() {
+                jobScheduleRing();
+            }
+        });
+        ringThread.setDaemon(true);
+        ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
+        ringThread.start();
+    }
 
-                while (!scheduleThreadToStop) {
+    private void jobScheduleStop() {
 
-                    // Scan Job
-                    long start = System.currentTimeMillis();
+        try {
+            TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
+        } catch (InterruptedException e) {
+            if (!scheduleThreadToStop) {
+                logger.error(e.getMessage(), e);
+            }
+        }
+        logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
 
-                    Connection conn = null;
-                    Boolean connAutoCommit = null;
-                    PreparedStatement preparedStatement = null;
+        // pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
+        int preReadCount = (xxlJobAdminConfig.getTriggerPoolFastMax() + xxlJobAdminConfig.getTriggerPoolSlowMax()) * 20;
 
-                    boolean preReadSuc = true;
-                    try {
+        while (!scheduleThreadToStop) {
 
-                        conn = dataSource.getConnection();
-                        connAutoCommit = conn.getAutoCommit();
-                        conn.setAutoCommit(false);
+            // Scan Job
+            long start = System.currentTimeMillis();
 
-                        preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
-                        preparedStatement.execute();
+            Connection conn = null;
+            boolean connAutoCommit = true;
+            PreparedStatement preparedStatement = null;
 
-                        // tx start
+            boolean preReadSuc = true;
+            try {
 
-                        // 1、pre read
-                        long nowTime = System.currentTimeMillis();
-                        List<XxlJobInfo> scheduleList = xxlJobInfoDao.scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
-                        if (scheduleList != null && scheduleList.size() > 0) {
-                            // 2、push time-ring
-                            for (XxlJobInfo jobInfo : scheduleList) {
+                conn = dataSource.getConnection();
+                connAutoCommit = conn.getAutoCommit();
+                conn.setAutoCommit(false);
 
-                                // time-ring jump
-                                if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
-                                    // 2.1、trigger-expire > 5s：pass && make next-trigger-time
-                                    logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
+                preparedStatement = conn.prepareStatement("select * from xxl_job_lock where lock_name = 'schedule_lock' for update");
+                preparedStatement.execute();
 
-                                    // fresh next
-                                    refreshNextValidTime(jobInfo, new Date());
+                // tx start
 
-                                } else if (nowTime > jobInfo.getTriggerNextTime()) {
-                                    // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
+                // 1、pre read
+                long nowTime = System.currentTimeMillis();
+                List<XxlJobInfo> scheduleList = xxlJobInfoDao.scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
+                if (scheduleList != null && scheduleList.size() > 0) {
+                    // 2、push time-ring
+                    for (XxlJobInfo jobInfo : scheduleList) {
 
-                                    // 1、trigger
-                                    jobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
-                                    logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
+                        // time-ring jump
+                        if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
+                            // 2.1、trigger-expire > 5s：pass && make next-trigger-time
+                            logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
 
-                                    // 2、fresh next
-                                    refreshNextValidTime(jobInfo, new Date());
+                            // fresh next
+                            refreshNextValidTime(jobInfo, new Date());
 
-                                    // next-trigger-time in 5s, pre-read again
-                                    if (jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
+                        } else if (nowTime > jobInfo.getTriggerNextTime()) {
+                            // 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
 
-                                        // 1、make ring second
-                                        int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
+                            // 1、trigger
+                            jobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null);
+                            logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
 
-                                        // 2、push time ring
-                                        pushTimeRing(ringSecond, jobInfo.getId());
+                            // 2、fresh next
+                            refreshNextValidTime(jobInfo, new Date());
 
-                                        // 3、fresh next
-                                        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
+                            // next-trigger-time in 5s, pre-read again
+                            if (jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
 
-                                    }
+                                // 1、make ring second
+                                int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
 
-                                } else {
-                                    // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
+                                // 2、push time ring
+                                pushTimeRing(ringSecond, jobInfo.getId());
 
-                                    // 1、make ring second
-                                    int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
+                                // 3、fresh next
+                                refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
 
-                                    // 2、push time ring
-                                    pushTimeRing(ringSecond, jobInfo.getId());
+                            }
 
-                                    // 3、fresh next
-                                    refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
+                        } else {
+                            // 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
 
-                                }
+                            // 1、make ring second
+                            int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
 
-                            }
+                            // 2、push time ring
+                            pushTimeRing(ringSecond, jobInfo.getId());
 
-                            // 3、update trigger info
-                            for (XxlJobInfo jobInfo : scheduleList) {
-                                xxlJobInfoDao.scheduleUpdate(jobInfo);
-                            }
+                            // 3、fresh next
+                            refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
 
-                        } else {
-                            preReadSuc = false;
                         }
 
-                        // tx stop
+                    }
+
+                    // 3、update trigger info
+                    for (XxlJobInfo jobInfo : scheduleList) {
+                        xxlJobInfoDao.scheduleUpdate(jobInfo);
+                    }
+
+                } else {
+                    preReadSuc = false;
+                }
 
+                // tx stop
+
+
+            } catch (Exception e) {
+                if (!scheduleThreadToStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e.getMessage());
+                }
+            } finally {
 
-                    } catch (Exception e) {
+                // commit
+                if (conn != null) {
+                    try {
+                        conn.commit();
+                    } catch (SQLException e) {
                         if (!scheduleThreadToStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
+                            logger.error(e.getMessage(), e);
                         }
-                    } finally {
-
-                        // commit
-                        if (conn != null) {
-                            try {
-                                conn.commit();
-                            } catch (SQLException e) {
-                                if (!scheduleThreadToStop) {
-                                    logger.error(e.getMessage(), e);
-                                }
-                            }
-                            try {
-                                conn.setAutoCommit(connAutoCommit);
-                            } catch (SQLException e) {
-                                if (!scheduleThreadToStop) {
-                                    logger.error(e.getMessage(), e);
-                                }
-                            }
-                            try {
-                                conn.close();
-                            } catch (SQLException e) {
-                                if (!scheduleThreadToStop) {
-                                    logger.error(e.getMessage(), e);
-                                }
-                            }
+                    }
+                    try {
+                        conn.setAutoCommit(connAutoCommit);
+                    } catch (SQLException e) {
+                        if (!scheduleThreadToStop) {
+                            logger.error(e.getMessage(), e);
                         }
-
-                        // close PreparedStatement
-                        if (null != preparedStatement) {
-                            try {
-                                preparedStatement.close();
-                            } catch (SQLException e) {
-                                if (!scheduleThreadToStop) {
-                                    logger.error(e.getMessage(), e);
-                                }
-                            }
+                    }
+                    try {
+                        conn.close();
+                    } catch (SQLException e) {
+                        if (!scheduleThreadToStop) {
+                            logger.error(e.getMessage(), e);
                         }
                     }
-                    long cost = System.currentTimeMillis() - start;
-
+                }
 
-                    // Wait seconds, align second
-                    if (cost < 1000) {  // scan-overtime, not wait
-                        try {
-                            // pre-read period: success > scan each second; fail > skip this period;
-                            TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
-                        } catch (InterruptedException e) {
-                            if (!scheduleThreadToStop) {
-                                logger.error(e.getMessage(), e);
-                            }
+                // close PreparedStatement
+                if (null != preparedStatement) {
+                    try {
+                        preparedStatement.close();
+                    } catch (SQLException e) {
+                        if (!scheduleThreadToStop) {
+                            logger.error(e.getMessage(), e);
                         }
                     }
-
                 }
-
-                logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
             }
-        });
-        scheduleThread.setDaemon(true);
-        scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
-        scheduleThread.start();
+            long cost = System.currentTimeMillis() - start;
 
 
-        // ring thread
-        ringThread = new Thread(new Runnable() {
-            @Override
-            public void run() {
-
-                // align second
+            // Wait seconds, align second
+            if (cost < 1000) {  // scan-overtime, not wait
                 try {
-                    TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
+                    // pre-read period: success > scan each second; fail > skip this period;
+                    TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
                 } catch (InterruptedException e) {
-                    if (!ringThreadToStop) {
+                    if (!scheduleThreadToStop) {
                         logger.error(e.getMessage(), e);
                     }
                 }
+            }
 
-                while (!ringThreadToStop) {
+        }
 
-                    try {
-                        // second data
-                        List<Integer> ringItemData = new ArrayList<>();
-                        int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
-                        for (int i = 0; i < 2; i++) {
-                            List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
-                            if (tmpData != null) {
-                                ringItemData.addAll(tmpData);
-                            }
-                        }
+        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
+    }
 
-                        // ring trigger
-                        logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData));
-                        if (ringItemData.size() > 0) {
-                            // do trigger
-                            for (int jobId : ringItemData) {
-                                // do trigger
-                                jobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
-                            }
-                            // clear
-                            ringItemData.clear();
-                        }
-                    } catch (Exception e) {
-                        if (!ringThreadToStop) {
-                            logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
-                        }
+    private void jobScheduleRing() {
+
+        // align second
+        try {
+            TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
+        } catch (InterruptedException e) {
+            if (!ringThreadToStop) {
+                logger.error(e.getMessage(), e);
+            }
+        }
+
+        while (!ringThreadToStop) {
+
+            try {
+                // second data
+                List<Integer> ringItemData = new ArrayList<>();
+                int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
+                for (int i = 0; i < 2; i++) {
+                    List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);
+                    if (tmpData != null) {
+                        ringItemData.addAll(tmpData);
                     }
+                }
 
-                    // next second, align second
-                    try {
-                        TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
-                    } catch (InterruptedException e) {
-                        if (!ringThreadToStop) {
-                            logger.error(e.getMessage(), e);
-                        }
+                // ring trigger
+                logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData));
+                if (ringItemData.size() > 0) {
+                    // do trigger
+                    for (int jobId : ringItemData) {
+                        // do trigger
+                        jobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
                     }
+                    // clear
+                    ringItemData.clear();
+                }
+            } catch (Exception e) {
+                if (!ringThreadToStop) {
+                    logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e.getMessage());
                 }
-                logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
             }
-        });
-        ringThread.setDaemon(true);
-        ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
-        ringThread.start();
+
+            // next second, align second
+            try {
+                TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
+            } catch (InterruptedException e) {
+                if (!ringThreadToStop) {
+                    logger.error(e.getMessage(), e);
+                }
+            }
+        }
+        logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
     }
 
     private void refreshNextValidTime(XxlJobInfo jobInfo, Date fromTime) throws ParseException {
@@ -302,14 +309,10 @@ public class JobScheduleHelper {
 
     private void pushTimeRing(int ringSecond, int jobId) {
         // push async ring
-        List<Integer> ringItemData = ringData.get(ringSecond);
-        if (ringItemData == null) {
-            ringItemData = new ArrayList<Integer>();
-            ringData.put(ringSecond, ringItemData);
-        }
+        List<Integer> ringItemData = ringData.computeIfAbsent(ringSecond, k -> new ArrayList<Integer>());
         ringItemData.add(jobId);
 
-        logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : " + ringSecond + " = " + Arrays.asList(ringItemData));
+        logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : {} = {}", ringSecond, ringItemData);
     }
 
     @PreDestroy

xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobTriggerPoolHelper.java

@@ -44,12 +44,13 @@ public class JobTriggerPoolHelper {
      */
     @PostConstruct
     public void start() {
+
         fastTriggerPool = new ThreadPoolExecutor(
                 10,
                 xxlJobAdminConfig.getTriggerPoolFastMax(),
                 60L,
                 TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>(1000),
+                new LinkedBlockingQueue<Runnable>(1024),
                 new ThreadFactory() {
                     @Override
                     public Thread newThread(Runnable r) {
@@ -62,7 +63,7 @@ public class JobTriggerPoolHelper {
                 xxlJobAdminConfig.getTriggerPoolSlowMax(),
                 60L,
                 TimeUnit.SECONDS,
-                new LinkedBlockingQueue<Runnable>(2000),
+                new LinkedBlockingQueue<Runnable>(2048),
                 new ThreadFactory() {
                     @Override
                     public Thread newThread(Runnable r) {
@@ -113,9 +114,9 @@ public class JobTriggerPoolHelper {
                 } finally {
 
                     // check timeout-count-map
-                    long minTim_now = System.currentTimeMillis() / 60000;
-                    if (minTim != minTim_now) {
-                        minTim = minTim_now;
+                    long minTimNow = System.currentTimeMillis() / 60000;
+                    if (minTim != minTimNow) {
+                        minTim = minTimNow;
                         jobTimeoutCountMap.clear();
                     }
 

xxl-job-admin/src/main/java/com/xxl/job/admin/core/util/JacksonUtil.java

@@ -1,6 +1,5 @@
 package com.xxl.job.admin.core.util;
 
-import com.fasterxml.jackson.core.JsonGenerationException;
 import com.fasterxml.jackson.core.JsonParseException;
 import com.fasterxml.jackson.databind.JavaType;
 import com.fasterxml.jackson.databind.JsonMappingException;
@@ -37,10 +36,6 @@ public class JacksonUtil {
     public static String writeValueAsString(Object obj) {
         try {
             return getInstance().writeValueAsString(obj);
-        } catch (JsonGenerationException e) {
-            logger.error(e.getMessage(), e);
-        } catch (JsonMappingException e) {
-            logger.error(e.getMessage(), e);
         } catch (IOException e) {
             logger.error(e.getMessage(), e);
         }
@@ -53,7 +48,6 @@ public class JacksonUtil {
      * @param jsonStr
      * @param clazz
      * @return obj
-     * @throws Exception
      */
     public static <T> T readValue(String jsonStr, Class<T> clazz) {
         try {

xxl-job-admin/src/main/java/com/xxl/job/admin/service/impl/AlarmServiceImpl.java

@@ -22,6 +22,8 @@ import java.util.Arrays;
 import java.util.Date;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
 
 /**
  * @author VcKerry on 2020/1/6
@@ -32,6 +34,10 @@ public class AlarmServiceImpl implements AlarmService {
 
     private static Logger logger = LoggerFactory.getLogger(AlarmServiceImpl.class);
 
+    /**
+     * email check
+     */
+    private final Pattern emailPattern = Pattern.compile("^([a-zA-Z0-9]*[-_]?[a-zA-Z0-9]+)*@([a-zA-Z0-9]*[-_]?[a-zA-Z0-9]+)+[.][A-Za-z]{2,3}([.][A-Za-z]{2})?$");
 
     @Value("${spring.mail.username}")
     private String emailUserName;
@@ -59,17 +65,24 @@ public class AlarmServiceImpl implements AlarmService {
 
         Set<String> alarmAddressSet = new HashSet<String>(Arrays.asList(info.getAlarmEmail().split(",")));
 
-        for (String email : alarmAddressSet) {
+        for (String val : alarmAddressSet) {
             try {
-                email = email.trim();
-                if (email.startsWith("http://") || email.startsWith("https://")) {
+                val = val.trim();
+                if (val.startsWith("http://") || val.startsWith("https://")) {
                     final String content = textContent(group, info, jobLog, personal);
-                    alarmResult = failAlarmHttp(email, title, content, jobLog.getId());
-                } else if (email.startsWith("sms://")) {
+                    alarmResult = failAlarmHttp(val, title, content, jobLog.getId());
+                } else if (val.startsWith("sms://")) {
                     // sms 短信
                 } else {
-                    String content = emailContent(group, info, jobLog);
-                    alarmResult = failAlarmEmail(email, title, content, jobLog.getId(), personal);
+                    final Matcher matcher = emailPattern.matcher(val);
+                    if (matcher.matches()) {
+                        // email...
+                        String content = emailContent(group, info, jobLog);
+                        alarmResult = failAlarmEmail(val, title, content, jobLog.getId(), personal);
+                    } else {
+                        // other type...
+                        logger.info("unknown alarm notify type. title: {}. jobId: {}", title, jobLog.getId());
+                    }
                 }
             } catch (Exception e) {
                 alarmResult = false;