From e7bea5e221c5b4938e8a7eda2443c37208e0ee58 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=86=B7=E5=86=B7?= <2270033969@qq.com> Date: Mon, 10 Mar 2025 18:58:46 +0800 Subject: [PATCH] =?UTF-8?q?refactor(xxl-job):=20=E4=BC=98=E5=8C=96?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=E6=B3=A8=E5=86=8C=E4=B8=AD=E5=BF=83=E7=9A=84?= =?UTF-8?q?=E6=97=B6=E9=97=B4=E5=A4=84=E7=90=86=E9=80=BB=E8=BE=91=EF=BC=8C?= =?UTF-8?q?=E5=9C=A8=20`XxlJobRegistryMapper.xml`=20=E4=B8=AD=EF=BC=8C?= =?UTF-8?q?=E7=AE=80=E5=8C=96=20SQL=20=E6=9F=A5=E8=AF=A2=E6=9D=A1=E4=BB=B6?= =?UTF-8?q?=EF=BC=8C=E7=9B=B4=E6=8E=A5=E4=BD=BF=E7=94=A8=20`expiredTime`?= =?UTF-8?q?=20=E5=8F=82=E6=95=B0=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../admin/controller/JobGroupController.java | 7 +- .../admin/core/thread/JobRegistryHelper.java | 342 +++++++++--------- .../xxl/job/admin/dao/XxlJobRegistryDao.java | 4 +- .../mybatis-mapper/XxlJobRegistryMapper.xml | 4 +- 4 files changed, 182 insertions(+), 175 deletions(-) diff --git a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java index 6b4394b9c..07fc04f83 100644 --- a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java +++ b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java @@ -17,6 +17,8 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.ResponseBody; +import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.*; /** @@ -161,7 +163,10 @@ public class JobGroupController { private List findRegistryByAppName(String appnameParam) { HashMap> appAddressMap = new HashMap>(); - List list = xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT, new Date()); + Instant now = Instant.now(); + Instant expiredInstant = now.minus(RegistryConfig.DEAD_TIMEOUT, ChronoUnit.SECONDS); // 假设 timeout 单位是秒 + Date expiredTime = Date.from(expiredInstant); + List list = xxlJobRegistryDao.findAll(expiredTime); if (list != null) { for (XxlJobRegistry item : list) { if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { diff --git a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java index 967b4d0af..7be954ae2 100644 --- a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java +++ b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java @@ -10,6 +10,8 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.StringUtils; +import java.time.Instant; +import java.time.temporal.ChronoUnit; import java.util.*; import java.util.concurrent.*; @@ -20,202 +22,202 @@ import java.util.concurrent.*; */ public class JobRegistryHelper { - private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class); + private static Logger logger = LoggerFactory.getLogger(JobRegistryHelper.class); - private static JobRegistryHelper instance = new JobRegistryHelper(); + private static JobRegistryHelper instance = new JobRegistryHelper(); - public static JobRegistryHelper getInstance() { - return instance; - } + public static JobRegistryHelper getInstance() { + return instance; + } - private ThreadPoolExecutor registryOrRemoveThreadPool = null; + private ThreadPoolExecutor registryOrRemoveThreadPool = null; - private Thread registryMonitorThread; + private Thread registryMonitorThread; - private volatile boolean toStop = false; + private volatile boolean toStop = false; - public void start() { + public void start() { - // for registry or remove - registryOrRemoveThreadPool = new ThreadPoolExecutor(2, 10, 30L, TimeUnit.SECONDS, - new LinkedBlockingQueue(2000), new ThreadFactory() { - @Override - public Thread newThread(Runnable r) { - return new Thread(r, - "xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-" + r.hashCode()); - } - }, new RejectedExecutionHandler() { - @Override - public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { - r.run(); - logger.warn( - ">>>>>>>>>>> xxl-job, registry or remove too fast, match threadpool rejected handler(run now)."); - } - }); + // for registry or remove + registryOrRemoveThreadPool = new ThreadPoolExecutor(2, 10, 30L, TimeUnit.SECONDS, + new LinkedBlockingQueue(2000), new ThreadFactory() { + @Override + public Thread newThread(Runnable r) { + return new Thread(r, + "xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-" + r.hashCode()); + } + }, new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + r.run(); + logger.warn( + ">>>>>>>>>>> xxl-job, registry or remove too fast, match threadpool rejected handler(run now)."); + } + }); - // for monitor - registryMonitorThread = new Thread(new Runnable() { - @Override - public void run() { - while (!toStop) { - try { - // auto registry group - List groupList = XxlJobAdminConfig.getAdminConfig() - .getXxlJobGroupDao() - .findByAddressType(0); - if (groupList != null && !groupList.isEmpty()) { + // for monitor + registryMonitorThread = new Thread(new Runnable() { + @Override + public void run() { + while (!toStop) { + try { + // auto registry group + List groupList = XxlJobAdminConfig.getAdminConfig() + .getXxlJobGroupDao() + .findByAddressType(0); + if (groupList != null && !groupList.isEmpty()) { - // remove dead address (admin/executor) - List ids = XxlJobAdminConfig.getAdminConfig() - .getXxlJobRegistryDao() - .findDead(RegistryConfig.DEAD_TIMEOUT, new Date()); - if (ids != null && ids.size() > 0) { - XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids); - } + // remove dead address (admin/executor) + Instant now = Instant.now(); + Instant expiredInstant = now.minus(RegistryConfig.DEAD_TIMEOUT, ChronoUnit.SECONDS); // 假设 timeout 单位是秒 + Date expiredTime = Date.from(expiredInstant); + List ids = XxlJobAdminConfig.getAdminConfig() + .getXxlJobRegistryDao() + .findDead(expiredTime); + if (ids != null && ids.size() > 0) { + XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids); + } - // fresh online address (admin/executor) - HashMap> appAddressMap = new HashMap>(); - List list = XxlJobAdminConfig.getAdminConfig() - .getXxlJobRegistryDao() - .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 registryList = appAddressMap.get(appname); - if (registryList == null) { - registryList = new ArrayList(); - } + // fresh online address (admin/executor) + HashMap> appAddressMap = new HashMap>(); + List list = XxlJobAdminConfig.getAdminConfig() + .getXxlJobRegistryDao() + .findAll(expiredTime); + if (list != null) { + for (XxlJobRegistry item : list) { + if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) { + String appname = item.getRegistryKey(); + List registryList = appAddressMap.get(appname); + if (registryList == null) { + registryList = new ArrayList(); + } - if (!registryList.contains(item.getRegistryValue())) { - registryList.add(item.getRegistryValue()); - } - appAddressMap.put(appname, registryList); - } - } - } + if (!registryList.contains(item.getRegistryValue())) { + registryList.add(item.getRegistryValue()); + } + appAddressMap.put(appname, registryList); + } + } + } - // fresh group address - for (XxlJobGroup group : groupList) { - List registryList = appAddressMap.get(group.getAppname()); - String addressListStr = null; - if (registryList != null && !registryList.isEmpty()) { - Collections.sort(registryList); - StringBuilder addressListSB = new StringBuilder(); - for (String item : registryList) { - addressListSB.append(item).append(","); - } - addressListStr = addressListSB.toString(); - addressListStr = addressListStr.substring(0, addressListStr.length() - 1); - } - group.setAddressList(addressListStr); - group.setUpdateTime(new Date()); + // fresh group address + for (XxlJobGroup group : groupList) { + List registryList = appAddressMap.get(group.getAppname()); + String addressListStr = null; + if (registryList != null && !registryList.isEmpty()) { + Collections.sort(registryList); + StringBuilder addressListSB = new StringBuilder(); + for (String item : registryList) { + addressListSB.append(item).append(","); + } + addressListStr = addressListSB.toString(); + addressListStr = addressListStr.substring(0, addressListStr.length() - 1); + } + group.setAddressList(addressListStr); + group.setUpdateTime(new Date()); - XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group); - } - } - } - 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); - } - } - } - logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop"); - } - }); - registryMonitorThread.setDaemon(true); - registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread"); - registryMonitorThread.start(); - } + XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group); + } + } + } 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); + } + } + } + logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop"); + } + }); + registryMonitorThread.setDaemon(true); + registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread"); + registryMonitorThread.start(); + } - public void toStop() { - toStop = true; + public void toStop() { + toStop = true; - // stop registryOrRemoveThreadPool - registryOrRemoveThreadPool.shutdownNow(); + // stop registryOrRemoveThreadPool + registryOrRemoveThreadPool.shutdownNow(); - // stop monitir (interrupt and wait) - registryMonitorThread.interrupt(); - try { - registryMonitorThread.join(); - } - catch (InterruptedException e) { - logger.error(e.getMessage(), e); - } - } + // stop monitir (interrupt and wait) + registryMonitorThread.interrupt(); + try { + registryMonitorThread.join(); + } catch (InterruptedException e) { + logger.error(e.getMessage(), e); + } + } - // ---------------------- helper ---------------------- + // ---------------------- helper ---------------------- - public ReturnT registry(RegistryParam registryParam) { + public ReturnT registry(RegistryParam registryParam) { - // valid - if (!StringUtils.hasText(registryParam.getRegistryGroup()) - || !StringUtils.hasText(registryParam.getRegistryKey()) - || !StringUtils.hasText(registryParam.getRegistryValue())) { - return new ReturnT(ReturnT.FAIL_CODE, "Illegal Argument."); - } + // valid + if (!StringUtils.hasText(registryParam.getRegistryGroup()) + || !StringUtils.hasText(registryParam.getRegistryKey()) + || !StringUtils.hasText(registryParam.getRegistryValue())) { + return new ReturnT(ReturnT.FAIL_CODE, "Illegal Argument."); + } - // async execute - registryOrRemoveThreadPool.execute(new Runnable() { - @Override - public void run() { - int ret = XxlJobAdminConfig.getAdminConfig() - .getXxlJobRegistryDao() - .registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), - registryParam.getRegistryValue(), new Date()); - if (ret < 1) { - XxlJobAdminConfig.getAdminConfig() - .getXxlJobRegistryDao() - .registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), - registryParam.getRegistryValue(), new Date()); + // async execute + registryOrRemoveThreadPool.execute(new Runnable() { + @Override + public void run() { + int ret = XxlJobAdminConfig.getAdminConfig() + .getXxlJobRegistryDao() + .registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), + registryParam.getRegistryValue(), new Date()); + if (ret < 1) { + XxlJobAdminConfig.getAdminConfig() + .getXxlJobRegistryDao() + .registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), + registryParam.getRegistryValue(), new Date()); - // fresh - freshGroupRegistryInfo(registryParam); - } - } - }); + // fresh + freshGroupRegistryInfo(registryParam); + } + } + }); - return ReturnT.SUCCESS; - } + return ReturnT.SUCCESS; + } - public ReturnT registryRemove(RegistryParam registryParam) { + public ReturnT registryRemove(RegistryParam registryParam) { - // valid - if (!StringUtils.hasText(registryParam.getRegistryGroup()) - || !StringUtils.hasText(registryParam.getRegistryKey()) - || !StringUtils.hasText(registryParam.getRegistryValue())) { - return new ReturnT(ReturnT.FAIL_CODE, "Illegal Argument."); - } + // valid + if (!StringUtils.hasText(registryParam.getRegistryGroup()) + || !StringUtils.hasText(registryParam.getRegistryKey()) + || !StringUtils.hasText(registryParam.getRegistryValue())) { + return new ReturnT(ReturnT.FAIL_CODE, "Illegal Argument."); + } - // async execute - registryOrRemoveThreadPool.execute(new Runnable() { - @Override - public void run() { - int ret = XxlJobAdminConfig.getAdminConfig() - .getXxlJobRegistryDao() - .registryDelete(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), - registryParam.getRegistryValue()); - if (ret > 0) { - // fresh - freshGroupRegistryInfo(registryParam); - } - } - }); + // async execute + registryOrRemoveThreadPool.execute(new Runnable() { + @Override + public void run() { + int ret = XxlJobAdminConfig.getAdminConfig() + .getXxlJobRegistryDao() + .registryDelete(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), + registryParam.getRegistryValue()); + if (ret > 0) { + // fresh + freshGroupRegistryInfo(registryParam); + } + } + }); - return ReturnT.SUCCESS; - } + return ReturnT.SUCCESS; + } - private void freshGroupRegistryInfo(RegistryParam registryParam) { - // Under consideration, prevent affecting core tables - } + private void freshGroupRegistryInfo(RegistryParam registryParam) { + // Under consideration, prevent affecting core tables + } } diff --git a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/dao/XxlJobRegistryDao.java b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/dao/XxlJobRegistryDao.java index 9b76add12..26e0a25bc 100644 --- a/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/dao/XxlJobRegistryDao.java +++ b/pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/dao/XxlJobRegistryDao.java @@ -13,11 +13,11 @@ import java.util.List; @Mapper public interface XxlJobRegistryDao { - public List findDead(@Param("timeout") int timeout, @Param("nowTime") Date nowTime); + public List findDead(@Param("expiredTime") Date expiredTime); public int removeDead(@Param("ids") List ids); - public List findAll(@Param("timeout") int timeout, @Param("nowTime") Date nowTime); + public List findAll(@Param("expiredTime") Date expiredTime); public int registryUpdate(@Param("registryGroup") String registryGroup, @Param("registryKey") String registryKey, @Param("registryValue") String registryValue, @Param("updateTime") Date updateTime); diff --git a/pigx-visual/pigx-xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml b/pigx-visual/pigx-xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml index 4898997ae..1d895db1f 100644 --- a/pigx-visual/pigx-xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml +++ b/pigx-visual/pigx-xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml @@ -22,8 +22,7 @@ @@ -37,6 +36,7 @@