- 将单例模式改为实例方法调用，通过getInstance()获取执行器实例 - 引入Helper类替换原有静态线程实现，包括ExecutorRegistryThreadHelper、 JobLogFileCleanThreadHelper和TriggerCallbackThreadHelper - 移除废弃的常量ELEGANT_SHUTDOWN_WAITING_SECONDS，改用Const类中的统一配置 - 更新所有内部方法调用为实例方法，增强代码的可测试性和灵活性 - 使用MessageQueue和CyclicThread优化回调消息处理机制 - 统一资源管理和生命周期控制，改进线程安全和资源清理逻辑3.4.1-release
xuxueli
3 days ago
parent
a7a3d44ee1
commit
9d4ba92537
@ -1,131 +0,0 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.constant.RegistType;
|
||||
import com.xxl.job.core.openapi.AdminBiz;
|
||||
import com.xxl.job.core.openapi.model.RegistryRequest;
|
||||
import com.xxl.job.core.constant.Const;
|
||||
import com.xxl.job.core.executor.XxlJobExecutor;
|
||||
import com.xxl.tool.response.Response;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* Created by xuxueli on 17/3/2.
|
||||
*/
|
||||
public class ExecutorRegistryThread {
|
||||
private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class);
|
||||
|
||||
private static ExecutorRegistryThread instance = new ExecutorRegistryThread();
|
||||
public static ExecutorRegistryThread getInstance(){
|
||||
return instance;
|
||||
}
|
||||
|
||||
private Thread registryThread;
|
||||
private volatile boolean toStop = false;
|
||||
public void start(final String appname, final String address){
|
||||
|
||||
// valid
|
||||
if (appname==null || appname.trim().length()==0) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appname is null.");
|
||||
return;
|
||||
}
|
||||
if (XxlJobExecutor.getAdminBizList() == null) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
|
||||
return;
|
||||
}
|
||||
|
||||
toStop = false;
|
||||
registryThread = new Thread(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
// registry
|
||||
while (!toStop) {
|
||||
try {
|
||||
RegistryRequest registryParam = new RegistryRequest(RegistType.EXECUTOR.name(), appname, address);
|
||||
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> registryResult = adminBiz.registry(registryParam);
|
||||
if (registryResult!=null && registryResult.isSuccess()) {
|
||||
registryResult = Response.ofSuccess();
|
||||
logger.debug(">>>>>>>>>>> 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 (Throwable e) {
|
||||
logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
|
||||
}
|
||||
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
try {
|
||||
if (!toStop) {
|
||||
TimeUnit.SECONDS.sleep(Const.BEAT_TIMEOUT);
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// registry remove
|
||||
try {
|
||||
RegistryRequest registryParam = new RegistryRequest(RegistType.EXECUTOR.name(), appname, address);
|
||||
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> registryResult = adminBiz.registryRemove(registryParam);
|
||||
if (registryResult!=null && registryResult.isSuccess()) {
|
||||
registryResult = Response.ofSuccess();
|
||||
logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
|
||||
break;
|
||||
} else {
|
||||
logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
logger.info(">>>>>>>>>>> xxl-job, executor registry thread destroy.");
|
||||
|
||||
}
|
||||
});
|
||||
registryThread.setDaemon(true);
|
||||
registryThread.setName("xxl-job, executor ExecutorRegistryThread");
|
||||
registryThread.start();
|
||||
}
|
||||
|
||||
public void toStop() {
|
||||
toStop = true;
|
||||
|
||||
// interrupt and wait
|
||||
if (registryThread != null) {
|
||||
registryThread.interrupt();
|
||||
try {
|
||||
registryThread.join();
|
||||
} catch (Throwable e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,107 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.constant.RegistType;
|
||||
import com.xxl.job.core.openapi.AdminBiz;
|
||||
import com.xxl.job.core.openapi.model.RegistryRequest;
|
||||
import com.xxl.job.core.constant.Const;
|
||||
import com.xxl.job.core.executor.XxlJobExecutor;
|
||||
import com.xxl.tool.concurrent.CyclicThread;
|
||||
import com.xxl.tool.core.CollectionTool;
|
||||
import com.xxl.tool.core.StringTool;
|
||||
import com.xxl.tool.response.Response;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
/**
|
||||
* Created by xuxueli on 17/3/2.
|
||||
*/
|
||||
public class ExecutorRegistryThreadHelper {
|
||||
private static final Logger logger = LoggerFactory.getLogger(ExecutorRegistryThreadHelper.class);
|
||||
|
||||
|
||||
/**
|
||||
* registry thread
|
||||
*/
|
||||
private CyclicThread registryThread;
|
||||
|
||||
/**
|
||||
* start
|
||||
*/
|
||||
public void start(final XxlJobExecutor xxlJobExecutor){
|
||||
|
||||
/**
|
||||
* valid
|
||||
*/
|
||||
if (StringTool.isBlank(xxlJobExecutor.getAppname())) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appname is null.");
|
||||
return;
|
||||
}
|
||||
if (CollectionTool.isEmpty(xxlJobExecutor.getAdminBizList())) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* registry thread
|
||||
*/
|
||||
registryThread = new CyclicThread("ExecutorRegistryThread#registryThread", true, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
RegistryRequest registryParam = new RegistryRequest(RegistType.EXECUTOR.name(), xxlJobExecutor.getAppname(), xxlJobExecutor.getAddress());
|
||||
for (AdminBiz adminBiz: xxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> registryResult = adminBiz.registry(registryParam);
|
||||
if (registryResult!=null && registryResult.isSuccess()) {
|
||||
registryResult = Response.ofSuccess();
|
||||
logger.debug(">>>>>>>>>>> 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 (Throwable e) {
|
||||
logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
}, Const.BEAT_TIMEOUT * 1000L, true);
|
||||
registryThread.start();
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* stop
|
||||
*/
|
||||
public void stop(final XxlJobExecutor xxlJobExecutor) {
|
||||
|
||||
/**
|
||||
* 1、stop registryThread
|
||||
*/
|
||||
registryThread.stop();
|
||||
|
||||
/**
|
||||
* 2、registry remove
|
||||
*/
|
||||
registryRemove(xxlJobExecutor);
|
||||
}
|
||||
|
||||
private void registryRemove(final XxlJobExecutor xxlJobExecutor){
|
||||
RegistryRequest registryParam = new RegistryRequest(RegistType.EXECUTOR.name(), xxlJobExecutor.getAppname(), xxlJobExecutor.getAddress());
|
||||
for (AdminBiz adminBiz: xxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> registryResult = adminBiz.registryRemove(registryParam);
|
||||
if (registryResult!=null && registryResult.isSuccess()) {
|
||||
registryResult = Response.ofSuccess();
|
||||
logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
|
||||
break;
|
||||
} else {
|
||||
logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}, error:{}", registryParam, e.getMessage());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,128 +0,0 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||
import com.xxl.tool.core.DateTool;
|
||||
import com.xxl.tool.io.FileTool;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.Calendar;
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* job file clean thread
|
||||
*
|
||||
* @author xuxueli 2017-12-29 16:23:43
|
||||
*/
|
||||
public class JobLogFileCleanThread {
|
||||
private static Logger logger = LoggerFactory.getLogger(JobLogFileCleanThread.class);
|
||||
|
||||
private static JobLogFileCleanThread instance = new JobLogFileCleanThread();
|
||||
public static JobLogFileCleanThread getInstance(){
|
||||
return instance;
|
||||
}
|
||||
|
||||
private Thread localThread;
|
||||
private volatile boolean toStop = false;
|
||||
public void start(final long logRetentionDays){
|
||||
|
||||
// limit min value
|
||||
if (logRetentionDays < 3 ) {
|
||||
return; // effective only when logRetentionDays >= 3
|
||||
}
|
||||
|
||||
toStop = false;
|
||||
localThread = new Thread(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
while (!toStop) {
|
||||
try {
|
||||
// clean log dir, over logRetentionDays
|
||||
File[] childDirs = new File(XxlJobFileAppender.getLogPath()).listFiles();
|
||||
if (childDirs!=null && childDirs.length>0) {
|
||||
|
||||
// today
|
||||
Calendar todayCal = Calendar.getInstance();
|
||||
todayCal.set(Calendar.HOUR_OF_DAY,0);
|
||||
todayCal.set(Calendar.MINUTE,0);
|
||||
todayCal.set(Calendar.SECOND,0);
|
||||
todayCal.set(Calendar.MILLISECOND,0);
|
||||
|
||||
Date todayDate = todayCal.getTime();
|
||||
|
||||
// clean expired logfile
|
||||
for (File childFile: childDirs) {
|
||||
|
||||
// valid log-path: must be directory
|
||||
if (!childFile.isDirectory()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// valid day log-path: like "---/2017-12-25/639.log"
|
||||
if (!childFile.getName().contains("-")) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// parse create-day of file-path
|
||||
Date logFileCreateDate = null;
|
||||
try {
|
||||
logFileCreateDate = DateTool.parseDate(childFile.getName());
|
||||
} catch (Exception e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
if (logFileCreateDate == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// check expired
|
||||
Date expiredDate = DateTool.addDays(logFileCreateDate, logRetentionDays);
|
||||
if (todayDate.getTime() > expiredDate.getTime()) {
|
||||
// expired, remove all log of this day
|
||||
FileTool.delete(childFile);
|
||||
//FileUtil.deleteRecursively(childFile);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
TimeUnit.DAYS.sleep(1);
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.info(">>>>>>>>>>> xxl-job, executor JobLogFileCleanThread thread destroy.");
|
||||
|
||||
}
|
||||
});
|
||||
localThread.setDaemon(true);
|
||||
localThread.setName("xxl-job, executor JobLogFileCleanThread");
|
||||
localThread.start();
|
||||
}
|
||||
|
||||
public void toStop() {
|
||||
toStop = true;
|
||||
|
||||
if (localThread == null) {
|
||||
return;
|
||||
}
|
||||
|
||||
// interrupt and wait
|
||||
localThread.interrupt();
|
||||
try {
|
||||
localThread.join();
|
||||
} catch (InterruptedException e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,108 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||
import com.xxl.tool.concurrent.CyclicThread;
|
||||
import com.xxl.tool.core.DateTool;
|
||||
import com.xxl.tool.io.FileTool;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.util.Calendar;
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* job file clean thread
|
||||
*
|
||||
* @author xuxueli 2017-12-29 16:23:43
|
||||
*/
|
||||
public class JobLogFileCleanThreadHelper {
|
||||
private static final Logger logger = LoggerFactory.getLogger(JobLogFileCleanThreadHelper.class);
|
||||
|
||||
|
||||
/**
|
||||
* monitor thread
|
||||
*/
|
||||
private CyclicThread logFileCleanThread;
|
||||
|
||||
|
||||
/**
|
||||
* start
|
||||
*/
|
||||
public void start(final long logRetentionDays){
|
||||
|
||||
/**
|
||||
* limit min value
|
||||
*/
|
||||
if (logRetentionDays < 3 ) {
|
||||
return; // effective only when logRetentionDays >= 3
|
||||
}
|
||||
|
||||
/**
|
||||
* logFileCleanThread
|
||||
*/
|
||||
logFileCleanThread = new CyclicThread("JobLogFileCleanThreadHelper#logFileCleanThread", true, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
// clean log dir, over logRetentionDays
|
||||
File[] childDirs = new File(XxlJobFileAppender.getLogPath()).listFiles();
|
||||
if (childDirs!=null && childDirs.length>0) {
|
||||
|
||||
// today
|
||||
Calendar todayCal = Calendar.getInstance();
|
||||
todayCal.set(Calendar.HOUR_OF_DAY,0);
|
||||
todayCal.set(Calendar.MINUTE,0);
|
||||
todayCal.set(Calendar.SECOND,0);
|
||||
todayCal.set(Calendar.MILLISECOND,0);
|
||||
|
||||
Date todayDate = todayCal.getTime();
|
||||
|
||||
// clean expired logfile
|
||||
for (File childFile: childDirs) {
|
||||
|
||||
// valid log-path: must be directory
|
||||
if (!childFile.isDirectory()) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// valid day log-path: like "---/2017-12-25/639.log"
|
||||
if (!childFile.getName().contains("-")) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// parse create-day of file-path
|
||||
Date logFileCreateDate = null;
|
||||
try {
|
||||
logFileCreateDate = DateTool.parseDate(childFile.getName());
|
||||
} catch (Exception e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
if (logFileCreateDate == null) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// check expired
|
||||
Date expiredDate = DateTool.addDays(logFileCreateDate, logRetentionDays);
|
||||
if (todayDate.getTime() > expiredDate.getTime()) {
|
||||
// expired, remove all log of this day
|
||||
FileTool.delete(childFile);
|
||||
//FileUtil.deleteRecursively(childFile);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}, DateTool.MILLIS_PER_DAY, true);
|
||||
logFileCleanThread.start();
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* stop
|
||||
*/
|
||||
public void stop() {
|
||||
|
||||
// stop logFileCleanThread
|
||||
logFileCleanThread.stop();
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,305 +0,0 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.openapi.AdminBiz;
|
||||
import com.xxl.job.core.openapi.model.CallbackRequest;
|
||||
import com.xxl.job.core.context.XxlJobContext;
|
||||
import com.xxl.job.core.context.XxlJobHelper;
|
||||
import com.xxl.job.core.constant.Const;
|
||||
import com.xxl.job.core.executor.XxlJobExecutor;
|
||||
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||
import com.xxl.tool.core.ArrayTool;
|
||||
import com.xxl.tool.core.CollectionTool;
|
||||
import com.xxl.tool.core.StringTool;
|
||||
import com.xxl.tool.crypto.Md5Tool;
|
||||
import com.xxl.tool.json.GsonTool;
|
||||
import com.xxl.tool.io.FileTool;
|
||||
import com.xxl.tool.response.Response;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
/**
|
||||
* Trigger Callback Thread
|
||||
*
|
||||
* Created by xuxueli on 16/7/22.
|
||||
*/
|
||||
public class TriggerCallbackThread {
|
||||
private static final Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);
|
||||
|
||||
private static final TriggerCallbackThread instance = new TriggerCallbackThread();
|
||||
public static TriggerCallbackThread getInstance(){
|
||||
return instance;
|
||||
}
|
||||
|
||||
/**
|
||||
* job results callback queue
|
||||
*/
|
||||
private final LinkedBlockingQueue<CallbackRequest> callBackQueue = new LinkedBlockingQueue<>();
|
||||
public static void pushCallBack(CallbackRequest callback){
|
||||
getInstance().callBackQueue.add(callback);
|
||||
logger.debug(">>>>>>>>>>> xxl-job, push callback request, logId:{}", callback.getLogId());
|
||||
}
|
||||
|
||||
/**
|
||||
* callback thread
|
||||
*/
|
||||
private Thread triggerCallbackThread;
|
||||
private Thread triggerRetryCallbackThread;
|
||||
private volatile boolean toStop = false;
|
||||
public void start() {
|
||||
|
||||
// valid
|
||||
if (XxlJobExecutor.getAdminBizList() == null) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null.");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
toStop = false;
|
||||
|
||||
/**
|
||||
* trigger callback thread
|
||||
*/
|
||||
|
||||
triggerCallbackThread = new Thread(new Runnable() {
|
||||
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
// normal callback
|
||||
while(!toStop){
|
||||
try {
|
||||
CallbackRequest callback = getInstance().callBackQueue.take();
|
||||
if (callback != null) {
|
||||
|
||||
// collect callback data
|
||||
List<CallbackRequest> callbackParamList = new ArrayList<>();
|
||||
callbackParamList.add(callback); // add one element
|
||||
int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList); // drainTo other all elements
|
||||
|
||||
// do callback, will retry if error
|
||||
if (CollectionTool.isNotEmpty(callbackParamList)) {
|
||||
doCallback(callbackParamList);
|
||||
}
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// thead stop, callback lasttime
|
||||
try {
|
||||
// collect callback data
|
||||
List<CallbackRequest> callbackParamList = new ArrayList<>();
|
||||
int drainToNum = getInstance().callBackQueue.drainTo(callbackParamList);
|
||||
|
||||
// do callback
|
||||
if (CollectionTool.isNotEmpty(callbackParamList)) {
|
||||
doCallback(callbackParamList);
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
logger.info(">>>>>>>>>>> xxl-job, executor callback thread destroy.");
|
||||
|
||||
}
|
||||
});
|
||||
triggerCallbackThread.setDaemon(true);
|
||||
triggerCallbackThread.setName("xxl-job, executor TriggerCallbackThread");
|
||||
triggerCallbackThread.start();
|
||||
|
||||
|
||||
/**
|
||||
* callback fail retry thread
|
||||
*/
|
||||
triggerRetryCallbackThread = new Thread(new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
while(!toStop){
|
||||
try {
|
||||
retryFailCallbackFile();
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
|
||||
}
|
||||
try {
|
||||
TimeUnit.SECONDS.sleep(Const.BEAT_TIMEOUT);
|
||||
} catch (Throwable e) {
|
||||
if (!toStop) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
}
|
||||
logger.info(">>>>>>>>>>> xxl-job, executor retry callback thread destroy.");
|
||||
}
|
||||
});
|
||||
triggerRetryCallbackThread.setDaemon(true);
|
||||
triggerRetryCallbackThread.setName("xxl-job, executor TriggerRetryCallbackThread");
|
||||
triggerRetryCallbackThread.start();
|
||||
|
||||
}
|
||||
public void toStop(){
|
||||
toStop = true;
|
||||
// stop callback, interrupt and wait
|
||||
if (triggerCallbackThread != null) { // support empty admin address
|
||||
triggerCallbackThread.interrupt();
|
||||
try {
|
||||
triggerCallbackThread.join();
|
||||
} catch (Throwable e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
// stop retry, interrupt and wait
|
||||
if (triggerRetryCallbackThread != null) {
|
||||
triggerRetryCallbackThread.interrupt();
|
||||
try {
|
||||
triggerRetryCallbackThread.join();
|
||||
} catch (Throwable e) {
|
||||
logger.error(e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
/**
|
||||
* do callback, will retry if error
|
||||
*
|
||||
* @param callbackParamList callback param list
|
||||
*/
|
||||
private void doCallback(List<CallbackRequest> callbackParamList){
|
||||
boolean callbackRet = false;
|
||||
// callback, will retry if error
|
||||
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> callbackResult = adminBiz.callback(callbackParamList);
|
||||
if (callbackResult!=null && callbackResult.isSuccess()) {
|
||||
callbackLog(callbackParamList, "<br>----------- xxl-job job callback finish.");
|
||||
callbackRet = true;
|
||||
break;
|
||||
} else {
|
||||
callbackLog(callbackParamList, "<br>----------- xxl-job job callback fail, callbackResult:" + callbackResult);
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
callbackLog(callbackParamList, "<br>----------- xxl-job job callback error, errorMsg:" + e.getMessage());
|
||||
}
|
||||
}
|
||||
if (!callbackRet) {
|
||||
appendFailCallbackFile(callbackParamList);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* callback log
|
||||
*/
|
||||
private void callbackLog(List<CallbackRequest> callbackParamList, String logContent){
|
||||
for (CallbackRequest callbackParam: callbackParamList) {
|
||||
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(callbackParam.getLogDateTim()), callbackParam.getLogId());
|
||||
XxlJobContext.setXxlJobContext(new XxlJobContext(
|
||||
-1,
|
||||
null,
|
||||
-1,
|
||||
-1,
|
||||
logFileName,
|
||||
-1,
|
||||
-1));
|
||||
XxlJobHelper.log(logContent);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ---------------------- fail-callback file ----------------------
|
||||
|
||||
/**
|
||||
* fail-callback file name
|
||||
*/
|
||||
private static final String failCallbackFileName = XxlJobFileAppender
|
||||
.getCallbackLogPath()
|
||||
.concat(File.separator)
|
||||
.concat("xxl-job-callback-{x}")
|
||||
.concat(".log");
|
||||
|
||||
/**
|
||||
* append fail-callback file
|
||||
*
|
||||
* @param callbackParamList callback param list
|
||||
*/
|
||||
private void appendFailCallbackFile(List<CallbackRequest> callbackParamList) {
|
||||
// valid
|
||||
if (CollectionTool.isEmpty(callbackParamList)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// generate callback data
|
||||
String callbackData = GsonTool.toJson(callbackParamList);
|
||||
String callbackDataMd5 = Md5Tool.md5(callbackData);
|
||||
|
||||
|
||||
// create file
|
||||
String finalLogFileName = failCallbackFileName.replace("{x}", callbackDataMd5);
|
||||
|
||||
// write callback log
|
||||
try {
|
||||
FileTool.writeString(finalLogFileName, callbackData);
|
||||
} catch (IOException e) {
|
||||
logger.error(">>>>>>>>>>> TriggerCallbackThread appendFailCallbackFile error, finalLogFileName:{}", finalLogFileName, e);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* retry fail-callback file
|
||||
*/
|
||||
private void retryFailCallbackFile() {
|
||||
|
||||
// valid
|
||||
File callbackLogPath = new File(XxlJobFileAppender.getCallbackLogPath());
|
||||
if (!callbackLogPath.exists()) {
|
||||
return;
|
||||
}
|
||||
// valid file type: must be directory
|
||||
if (!FileTool.isDirectory(callbackLogPath)) {
|
||||
FileTool.delete(callbackLogPath);
|
||||
return;
|
||||
}
|
||||
// valid file in path: pass if empty
|
||||
if (ArrayTool.isEmpty(callbackLogPath.listFiles())) {
|
||||
return;
|
||||
}
|
||||
|
||||
// load and clear file, do retry
|
||||
for (File callbackLogFile: callbackLogPath.listFiles()) {
|
||||
try {
|
||||
// load data
|
||||
String callbackData = FileTool.readString(callbackLogFile.getPath());
|
||||
if (StringTool.isBlank(callbackData)) {
|
||||
FileTool.delete(callbackLogFile);
|
||||
continue;
|
||||
}
|
||||
|
||||
// parse callback param
|
||||
List<CallbackRequest> callbackParamList = GsonTool.fromJsonList(callbackData, CallbackRequest.class);
|
||||
FileTool.delete(callbackLogFile);
|
||||
|
||||
// retry callback
|
||||
doCallback(callbackParamList);
|
||||
} catch (IOException e) {
|
||||
logger.error(">>>>>>>>>>> TriggerCallbackThread retryFailCallbackFile error, callbackLogFile:{}", callbackLogFile.getPath(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@ -0,0 +1,237 @@
|
||||
package com.xxl.job.core.thread;
|
||||
|
||||
import com.xxl.job.core.constant.Const;
|
||||
import com.xxl.job.core.context.XxlJobContext;
|
||||
import com.xxl.job.core.context.XxlJobHelper;
|
||||
import com.xxl.job.core.executor.XxlJobExecutor;
|
||||
import com.xxl.job.core.log.XxlJobFileAppender;
|
||||
import com.xxl.job.core.openapi.AdminBiz;
|
||||
import com.xxl.job.core.openapi.model.CallbackRequest;
|
||||
import com.xxl.tool.concurrent.CyclicThread;
|
||||
import com.xxl.tool.concurrent.MessageQueue;
|
||||
import com.xxl.tool.core.ArrayTool;
|
||||
import com.xxl.tool.core.CollectionTool;
|
||||
import com.xxl.tool.core.StringTool;
|
||||
import com.xxl.tool.crypto.Md5Tool;
|
||||
import com.xxl.tool.io.FileTool;
|
||||
import com.xxl.tool.json.GsonTool;
|
||||
import com.xxl.tool.response.Response;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
import java.io.File;
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collections;
|
||||
import java.util.Date;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Trigger Callback Thread
|
||||
*
|
||||
* Created by xuxueli on 16/7/22.
|
||||
*/
|
||||
public class TriggerCallbackThreadHelper {
|
||||
private static final Logger logger = LoggerFactory.getLogger(TriggerCallbackThreadHelper.class);
|
||||
|
||||
|
||||
/**
|
||||
* callback message-queue
|
||||
*/
|
||||
private volatile MessageQueue<CallbackRequest> callbackMessageQueue;
|
||||
|
||||
/**
|
||||
* retry callback-file thread
|
||||
*/
|
||||
private CyclicThread retryCallbackThread;
|
||||
|
||||
|
||||
/**
|
||||
* start
|
||||
*/
|
||||
public void start(final XxlJobExecutor xxlJobExecutor) {
|
||||
|
||||
// valid
|
||||
if (xxlJobExecutor.getAdminBizList() == null) {
|
||||
logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null.");
|
||||
return;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* 1、callback message-queue
|
||||
*/
|
||||
callbackMessageQueue = new MessageQueue<CallbackRequest>(
|
||||
"TriggerCallbackThreadHelper#callbackMessageQueue",
|
||||
messages -> {
|
||||
|
||||
// do callback
|
||||
doCallback(messages, xxlJobExecutor);
|
||||
},
|
||||
1,
|
||||
50);
|
||||
|
||||
|
||||
/**
|
||||
* 2、retry callback-file thread
|
||||
*/
|
||||
retryCallbackThread = new CyclicThread("TriggerCallbackThreadHelper#retryCallbackThread", true, new Runnable() {
|
||||
@Override
|
||||
public void run() {
|
||||
|
||||
// valid empty path
|
||||
File callbackLogPath = new File(XxlJobFileAppender.getCallbackLogPath());
|
||||
if (!callbackLogPath.exists()) {
|
||||
return;
|
||||
}
|
||||
// valid file type: must be directory
|
||||
if (!FileTool.isDirectory(callbackLogPath)) {
|
||||
FileTool.delete(callbackLogPath);
|
||||
return;
|
||||
}
|
||||
// valid none file
|
||||
if (ArrayTool.isEmpty(callbackLogPath.listFiles())) {
|
||||
return;
|
||||
}
|
||||
|
||||
// load and clear file, do retry
|
||||
for (File callbackLogFile: callbackLogPath.listFiles()) {
|
||||
try {
|
||||
// load data
|
||||
String callbackData = FileTool.readString(callbackLogFile.getPath());
|
||||
if (StringTool.isBlank(callbackData)) {
|
||||
FileTool.delete(callbackLogFile);
|
||||
continue;
|
||||
}
|
||||
|
||||
// parse callback param
|
||||
List<CallbackRequest> callbackParamList = GsonTool.fromJsonList(callbackData, CallbackRequest.class);
|
||||
FileTool.delete(callbackLogFile);
|
||||
|
||||
// retry callback
|
||||
doCallback(callbackParamList, xxlJobExecutor);
|
||||
} catch (IOException e) {
|
||||
logger.error(">>>>>>>>>>> TriggerCallbackThread retryFailCallbackFile error, callbackLogFile:{}", callbackLogFile.getPath(), e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}, Const.BEAT_TIMEOUT * 1000L, true);
|
||||
retryCallbackThread.start();
|
||||
}
|
||||
|
||||
/**
|
||||
* stop
|
||||
*/
|
||||
public void stop(){
|
||||
// 1、stop callbackMessageQueue
|
||||
if (callbackMessageQueue != null) {
|
||||
callbackMessageQueue.stop(); // attempt wait for callback finish
|
||||
}
|
||||
|
||||
// 2、stop retryCallbackThread
|
||||
retryCallbackThread.stop();
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* submit callback message
|
||||
*/
|
||||
public void pushCallBack(CallbackRequest callback){
|
||||
if (!callbackMessageQueue.produce(callback)) {
|
||||
doCallback(new ArrayList<>(Collections.singletonList(callback)), XxlJobExecutor.getInstance());
|
||||
}
|
||||
logger.debug(">>>>>>>>>>> xxl-job, push callback request, logId:{}", callback.getLogId());
|
||||
}
|
||||
|
||||
|
||||
// ---------------------- do callback ----------------------
|
||||
|
||||
/**
|
||||
* do callback, will retry if error
|
||||
*
|
||||
* @param callbackParamList callback param list
|
||||
*/
|
||||
private void doCallback(List<CallbackRequest> callbackParamList, final XxlJobExecutor xxlJobExecutor){
|
||||
boolean callbackRet = false;
|
||||
|
||||
// callback request, will retry + append-log if fail
|
||||
for (AdminBiz adminBiz: xxlJobExecutor.getAdminBizList()) {
|
||||
try {
|
||||
Response<String> callbackResult = adminBiz.callback(callbackParamList);
|
||||
if (callbackResult!=null && callbackResult.isSuccess()) {
|
||||
appendCallbackResult(callbackParamList, "<br>----------- xxl-job job callback finish.");
|
||||
callbackRet = true;
|
||||
break;
|
||||
} else {
|
||||
appendCallbackResult(callbackParamList, "<br>----------- xxl-job job callback fail, callbackResult:" + callbackResult);
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
appendCallbackResult(callbackParamList, "<br>----------- xxl-job job callback error, errorMsg:" + e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// write callback-file, will retry later
|
||||
if (!callbackRet) {
|
||||
writeCallbackLog(callbackParamList);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* append callback result, to each joblog
|
||||
*/
|
||||
private void appendCallbackResult(List<CallbackRequest> callbackParamList, String logContent){
|
||||
for (CallbackRequest callbackParam: callbackParamList) {
|
||||
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(callbackParam.getLogDateTim()), callbackParam.getLogId());
|
||||
XxlJobContext.setXxlJobContext(new XxlJobContext(
|
||||
-1,
|
||||
null,
|
||||
-1,
|
||||
-1,
|
||||
logFileName,
|
||||
-1,
|
||||
-1));
|
||||
XxlJobHelper.log(logContent);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
// ---------------------- fail-callback file ----------------------
|
||||
|
||||
/**
|
||||
* fail-callback file name
|
||||
*/
|
||||
private static final String failCallbackFileName = XxlJobFileAppender
|
||||
.getCallbackLogPath()
|
||||
.concat(File.separator)
|
||||
.concat("xxl-job-callback-{x}")
|
||||
.concat(".log");
|
||||
|
||||
/**
|
||||
* write fail-callback file, will retry later
|
||||
*
|
||||
* @param callbackParamList callback param list
|
||||
*/
|
||||
private void writeCallbackLog(List<CallbackRequest> callbackParamList) {
|
||||
// valid
|
||||
if (CollectionTool.isEmpty(callbackParamList)) {
|
||||
return;
|
||||
}
|
||||
|
||||
// generate callback data
|
||||
String callbackData = GsonTool.toJson(callbackParamList);
|
||||
String callbackDataMd5 = Md5Tool.md5(callbackData);
|
||||
|
||||
|
||||
// create file
|
||||
String finalLogFileName = failCallbackFileName.replace("{x}", callbackDataMd5);
|
||||
|
||||
// write callback log
|
||||
try {
|
||||
FileTool.writeString(finalLogFileName, callbackData);
|
||||
} catch (IOException e) {
|
||||
logger.error(">>>>>>>>>>> TriggerCallbackThread appendFailCallbackFile error, finalLogFileName:{}", finalLogFileName, e);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in new issue