Skip to content

X
XXL-JOB
提交 8e07d501 authored 作者: xuxueli's avatar xuxueli
浏览文件
操作

心跳注册和结果回调，重试逻辑优化

上级 ff8e43af
隐藏空白字符变更
内嵌 并排
正在显示 包含 33 行增加 和 27 行删除
xxl-job-core/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java
浏览文件 @ 8e07d501
...@@ -49,25 +49,31 @@ public class ExecutorRegistryThread extends Thread { ...@@ -49,25 +49,31 @@ public class ExecutorRegistryThread extends Thread {
@Override @Override
public void run() { public void run() {
while (!toStop) { while (!toStop) {
try { try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, executorAddress); RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appName, executorAddress);
ReturnT<String> registryResult = null;
for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) { for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) {
String apiUrl = addressUrl.concat("/api"); String apiUrl = addressUrl.concat("/api");
AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject(); try {
registryResult = adminBiz.registry(registryParam); AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject();
if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) { ReturnT<String> registryResult = adminBiz.registry(registryParam);
registryResult = ReturnT.SUCCESS; if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
break; registryResult = ReturnT.SUCCESS;
logger.info(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
break;
} else {
logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
}
} catch (Exception e) {
logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
} }
} }
logger.info(">>>>>>>>>>> xxl-job Executor registry {}, RegistryParam:{}, registryResult:{}",
new Object[]{(registryResult.getCode()==ReturnT.SUCCESS_CODE?"success":"fail"), registryParam.toString(), registryResult.toString()});
} catch (Exception e) { } catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job ExecutorRegistryThread Exception:", e); logger.error(e.getMessage(), e);
} }
try { try {
......
xxl-job-core/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
浏览文件 @ 8e07d501
...@@ -37,40 +37,40 @@ public class TriggerCallbackThread { ...@@ -37,40 +37,40 @@ public class TriggerCallbackThread {
HandleCallbackParam callback = getInstance().callBackQueue.take(); HandleCallbackParam callback = getInstance().callBackQueue.take();
if (callback != null) { if (callback != null) {
// valid
if (XxlJobExecutor.adminAddresses==null || XxlJobExecutor.adminAddresses.trim().length()==0) {
logger.warn(">>>>>>>>>>>> xxl-job callback fail, adminAddresses is null.");
continue;
}
// callback list param // callback list param
List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>(); List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList); int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
callbackParamList.add(callback); callbackParamList.add(callback);
// callback, will retry if error // valid
try { if (XxlJobExecutor.adminAddresses==null || XxlJobExecutor.adminAddresses.trim().length()==0) {
logger.warn(">>>>>>>>>>>> xxl-job callback fail, adminAddresses is null, callbackParamList：{}", callbackParamList);
continue;
}
ReturnT<String> callbackResult = null; // callback, will retry if error
for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) { for (String addressUrl: XxlJobExecutor.adminAddresses.split(",")) {
String apiUrl = addressUrl.concat("/api"); String apiUrl = addressUrl.concat("/api");
try {
AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject(); AdminBiz adminBiz = (AdminBiz) new NetComClientProxy(AdminBiz.class, apiUrl).getObject();
callbackResult = adminBiz.callback(callbackParamList); ReturnT<String> callbackResult = adminBiz.callback(callbackParamList);
if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) { if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
callbackResult = ReturnT.SUCCESS; callbackResult = ReturnT.SUCCESS;
logger.info(">>>>>>>>>>> xxl-job callback success, callbackParamList:{}, callbackResult:{}", new Object[]{callbackParamList, callbackResult});
break; break;
} else {
logger.info(">>>>>>>>>>> xxl-job callback fail, callbackParamList:{}, callbackResult:{}", new Object[]{callbackParamList, callbackResult});
} }
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job callback error, callbackParamList：{}", callbackParamList, e);
//getInstance().callBackQueue.addAll(callbackParamList);
} }
logger.info(">>>>>>>>>>> xxl-job callback, callbackParamList:{}, callbackResult:{}", new Object[]{callbackParamList, callbackResult});
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job TriggerCallbackThread Exception:", e);
//getInstance().callBackQueue.addAll(callbackParamList);
} }
} }
} catch (Exception e) { } catch (Exception e) {
logger.error("", e); logger.error(e.getMessage(), e);
} }
} }
} }
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论