zianxin/pigx
1
0 Fork 0
mirror of http://git.pig4cloud.com/pig/pigx.git synced 2026-11-08 20:49:00 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(xxl-job): 优化任务注册中心的时间处理逻辑，在 XxlJobRegistryMapper.xml 中，简化 SQL 查询条件，直接使用 expiredTime 参数。

Browse Source
This commit is contained in:
冷冷 committed 2025-03-10 18:58:46 +08:00
1 parent 186505deac
commit e7bea5e221
4 files changed
+182 -175
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/controller/JobGroupController.java
+6 -1
View File
@@ -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<String> findRegistryByAppName(String appnameParam) {
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
List<XxlJobRegistry> 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<XxlJobRegistry> list = xxlJobRegistryDao.findAll(expiredTime);
if (list != null) {
for (XxlJobRegistry item : list) {
if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/core/thread/JobRegistryHelper.java
+172 -170
View File
@@ -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<Runnable>(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<Runnable>(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<XxlJobGroup> 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<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig()
.getXxlJobGroupDao()
.findByAddressType(0);
if (groupList != null && !groupList.isEmpty()) {
// remove dead address (admin/executor)
List<Integer> 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<Integer> ids = XxlJobAdminConfig.getAdminConfig()
.getXxlJobRegistryDao()
.findDead(expiredTime);
if (ids != null && ids.size() > 0) {
XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
}
// fresh online address (admin/executor)
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
List<XxlJobRegistry> 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<String> registryList = appAddressMap.get(appname);
if (registryList == null) {
registryList = new ArrayList<String>();
}
// fresh online address (admin/executor)
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
List<XxlJobRegistry> 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<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);
}
}
}
if (!registryList.contains(item.getRegistryValue())) {
registryList.add(item.getRegistryValue());
}
appAddressMap.put(appname, registryList);
}
}
}
// 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);
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<String> 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<String> registry(RegistryParam registryParam) {
public ReturnT<String> registry(RegistryParam registryParam) {
// valid
if (!StringUtils.hasText(registryParam.getRegistryGroup())
|| !StringUtils.hasText(registryParam.getRegistryKey())
|| !StringUtils.hasText(registryParam.getRegistryValue())) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
}
// valid
if (!StringUtils.hasText(registryParam.getRegistryGroup())
|| !StringUtils.hasText(registryParam.getRegistryKey())
|| !StringUtils.hasText(registryParam.getRegistryValue())) {
return new ReturnT<String>(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<String> registryRemove(RegistryParam registryParam) {
public ReturnT<String> registryRemove(RegistryParam registryParam) {
// valid
if (!StringUtils.hasText(registryParam.getRegistryGroup())
|| !StringUtils.hasText(registryParam.getRegistryKey())
|| !StringUtils.hasText(registryParam.getRegistryValue())) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
}
// valid
if (!StringUtils.hasText(registryParam.getRegistryGroup())
|| !StringUtils.hasText(registryParam.getRegistryKey())
|| !StringUtils.hasText(registryParam.getRegistryValue())) {
return new ReturnT<String>(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
}
}
pigx-visual/pigx-xxl-job-admin/src/main/java/com/xxl/job/admin/dao/XxlJobRegistryDao.java
+2 -2
View File
@@ -13,11 +13,11 @@ import java.util.List;
@Mapper
public interface XxlJobRegistryDao {
public List<Integer> findDead(@Param("timeout") int timeout, @Param("nowTime") Date nowTime);
public List<Integer> findDead(@Param("expiredTime") Date expiredTime);
public int removeDead(@Param("ids") List<Integer> ids);
public List<XxlJobRegistry> findAll(@Param("timeout") int timeout, @Param("nowTime") Date nowTime);
public List<XxlJobRegistry> findAll(@Param("expiredTime") Date expiredTime);
public int registryUpdate(@Param("registryGroup") String registryGroup, @Param("registryKey") String registryKey,
@Param("registryValue") String registryValue, @Param("updateTime") Date updateTime);
pigx-visual/pigx-xxl-job-admin/src/main/resources/mybatis-mapper/XxlJobRegistryMapper.xml
+2 -2
View File
@@ -22,8 +22,7 @@
<select id="findDead" parameterType="java.util.HashMap" resultType="java.lang.Integer" >
SELECT t.id
FROM xxl_job_registry AS t
WHERE t.update_time <![CDATA[ < ]]> ( sysdate() -(CONCAT(#{timeout}, '.0')/24/60/60))
WHERE t.update_time <![CDATA[ < ]]> #{expiredTime}
</select>
<delete id="removeDead" parameterType="java.lang.Integer" >
@@ -37,6 +36,7 @@
<select id="findAll" parameterType="java.util.HashMap" resultMap="XxlJobRegistry">
SELECT <include refid="Base_Column_List" />
FROM xxl_job_registry AS t
WHERE t.update_time > #{expiredTime}
</select>
<update id="registryUpdate" >