初始代码

This commit is contained in:
wangmingwei
2026-04-21 17:18:06 +08:00
parent ac9e431f80
commit aa74bd741e
259 changed files with 45901 additions and 0 deletions

104
xxl-job-core/pom.xml Normal file
View File

@@ -0,0 +1,104 @@
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<artifactId>yunzhupaas-scheduletask-starter</artifactId>
<groupId>com.yunzhupaas</groupId>
<version>5.2.0-RELEASE</version>
</parent>
<artifactId>xxl-job-core</artifactId>
<packaging>jar</packaging>
<name>${project.artifactId}</name>
<description>A distributed task scheduling framework.</description>
<url>https://www.xuxueli.com/</url>
<properties>
<netty.version>4.1.115.Final</netty.version>
<gson.version>2.11.0</gson.version>
<groovy.version>4.0.24</groovy.version>
<slf4j-api.version>2.0.16</slf4j-api.version>
</properties>
<dependencies>
<!-- ********************** embed server: netty + gson ********************** -->
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
<version>${netty.version}</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>${gson.version}</version>
</dependency>
<!-- ********************** plugin ********************** -->
<!-- groovy-all -->
<dependency>
<groupId>org.apache.groovy</groupId>
<artifactId>groovy</artifactId>
<version>${groovy.version}</version>
</dependency>
<!-- spring-context -->
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
<scope>provided</scope>
</dependency>
<!-- ********************** base ********************** -->
<!-- slf4j -->
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>
</dependencies>
<profiles>
<profile>
<id>boot3</id>
<activation>
<jdk>[17,)</jdk>
</activation>
<properties>
<servlet.annotation-api.version>2.1.1</servlet.annotation-api.version>
</properties>
<dependencies>
<!-- jakarta.annotation-api -->
<dependency>
<groupId>jakarta.annotation</groupId>
<artifactId>jakarta.annotation-api</artifactId>
<version>${servlet.annotation-api.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
</profile>
<profile>
<id>boot2</id>
<activation>
<jdk>(,17)</jdk>
</activation>
<properties>
<servlet.annotation-api.version>1.3.2</servlet.annotation-api.version>
</properties>
<dependencies>
<!-- javax.annotation-api -->
<dependency>
<groupId>javax.annotation</groupId>
<artifactId>javax.annotation-api</artifactId>
<version>${servlet.annotation-api.version}</version>
<scope>provided</scope>
</dependency>
</dependencies>
</profile>
</profiles>
</project>

View File

@@ -0,0 +1,54 @@
package com.xxl.job.core.biz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryHandlerName;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import java.util.List;
/**
* @author xuxueli 2017-07-27 21:52:49
*/
public interface AdminBiz {
// ---------------------- callback ----------------------
/**
* callback
*
* @param callbackParamList
* @return
*/
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList);
// ---------------------- registry ----------------------
/**
* registry
*
* @param registryParam
* @return
*/
public ReturnT<String> registry(RegistryParam registryParam);
/**
* registry remove
*
* @param registryParam
* @return
*/
public ReturnT<String> registryRemove(RegistryParam registryParam);
/**
* registry handlerName
*
* @param registryHandlerName
* @return
*/
public ReturnT<String> registryHandlerName(RegistryHandlerName registryHandlerName);
// ---------------------- biz (custome) ----------------------
// group、job ... manage
}

View File

@@ -0,0 +1,46 @@
package com.xxl.job.core.biz;
import com.xxl.job.core.biz.model.*;
/**
* Created by xuxueli on 17/3/1.
*/
public interface ExecutorBiz {
/**
* beat
* @return
*/
public ReturnT<String> beat();
/**
* idle beat
*
* @param idleBeatParam
* @return
*/
public ReturnT<String> idleBeat(IdleBeatParam idleBeatParam);
/**
* run
* @param triggerParam
* @return
*/
public ReturnT<String> run(TriggerParam triggerParam);
/**
* kill
* @param killParam
* @return
*/
public ReturnT<String> kill(KillParam killParam);
/**
* log
* @param logParam
* @return
*/
public ReturnT<LogResult> log(LogParam logParam);
}

View File

@@ -0,0 +1,73 @@
package com.xxl.job.core.biz.client;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.RegistryHandlerName;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.util.XxlJobRemotingUtil;
import java.util.List;
/**
* admin api test
*
* @author xuxueli 2017-07-28 22:14:52
*/
public class AdminBizClient implements AdminBiz {
public AdminBizClient() {
}
public AdminBizClient(String addressUrl, String accessToken) {
this.addressUrl = addressUrl;
this.accessToken = accessToken;
// valid
if (!this.addressUrl.endsWith("/")) {
this.addressUrl = this.addressUrl + "/";
}
}
private String addressUrl;
private String accessToken;
private int timeout = 3;
@Override
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
return XxlJobRemotingUtil.postBody(addressUrl + "api/callback", accessToken, timeout, callbackParamList,
String.class);
}
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, timeout, registryParam,
String.class);
}
@Override
public ReturnT<String> registryRemove(RegistryParam registryParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "api/registryRemove", accessToken, timeout, registryParam,
String.class);
}
@Override
public ReturnT<String> registryHandlerName(RegistryHandlerName registryHandlerName) {
Thread thread = new Thread(() -> {
ReturnT<String> returnT = XxlJobRemotingUtil.postBody(
addressUrl + "api/" + registryHandlerName.getUri() + "/handlerName", accessToken, timeout,
registryHandlerName, String.class);
if (ReturnT.FAIL_CODE == returnT.getCode()) {
try {
Thread.sleep(3000);
} catch (InterruptedException e1) {
e1.printStackTrace();
}
registryHandlerName(registryHandlerName);
}
});
thread.start();
return null;
}
}

View File

@@ -0,0 +1,57 @@
package com.xxl.job.core.biz.client;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.*;
import com.xxl.job.core.util.XxlJobRemotingUtil;
/**
* admin api test
*
* @author xuxueli 2017-07-28 22:14:52
*/
public class ExecutorBizClient implements ExecutorBiz {
public ExecutorBizClient() {
}
public ExecutorBizClient(String addressUrl, String accessToken) {
this.addressUrl = addressUrl;
this.accessToken = accessToken;
// valid
if (!this.addressUrl.endsWith("/")) {
this.addressUrl = this.addressUrl + "/";
}
}
private String addressUrl ;
private String accessToken;
private int timeout = 3;
@Override
public ReturnT<String> beat() {
return XxlJobRemotingUtil.postBody(addressUrl+"beat", accessToken, timeout, "", String.class);
}
@Override
public ReturnT<String> idleBeat(IdleBeatParam idleBeatParam){
return XxlJobRemotingUtil.postBody(addressUrl+"idleBeat", accessToken, timeout, idleBeatParam, String.class);
}
@Override
public ReturnT<String> run(TriggerParam triggerParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class);
}
@Override
public ReturnT<String> kill(KillParam killParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "kill", accessToken, timeout, killParam, String.class);
}
@Override
public ReturnT<LogResult> log(LogParam logParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "log", accessToken, timeout, logParam, LogResult.class);
}
}

View File

@@ -0,0 +1,173 @@
package com.xxl.job.core.biz.impl;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.*;
import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.glue.GlueFactory;
import com.xxl.job.core.glue.GlueTypeEnum;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.impl.GlueJobHandler;
import com.xxl.job.core.handler.impl.ScriptJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.thread.JobThread;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.Date;
/**
* Created by xuxueli on 17/3/1.
*/
public class ExecutorBizImpl implements ExecutorBiz {
private static Logger logger = LoggerFactory.getLogger(ExecutorBizImpl.class);
@Override
public ReturnT<String> beat() {
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> idleBeat(IdleBeatParam idleBeatParam) {
// isRunningOrHasQueue
boolean isRunningOrHasQueue = false;
JobThread jobThread = XxlJobExecutor.loadJobThread(idleBeatParam.getJobId());
if (jobThread != null && jobThread.isRunningOrHasQueue()) {
isRunningOrHasQueue = true;
}
if (isRunningOrHasQueue) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue.");
}
return ReturnT.SUCCESS;
}
@Override
public ReturnT<String> run(TriggerParam triggerParam) {
// load oldjobHandler + jobThread
JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId());
IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null;
String removeOldReason = null;
// validjobHandler + jobThread
GlueTypeEnum glueTypeEnum = GlueTypeEnum.match(triggerParam.getGlueType());
if (GlueTypeEnum.BEAN == glueTypeEnum) {
// new jobhandler
IJobHandler newJobHandler = XxlJobExecutor.loadJobHandler(triggerParam.getExecutorHandler());
// valid old jobThread
if (jobThread!=null && jobHandler != newJobHandler) {
// change handler, need kill old thread
removeOldReason = "change jobhandler or glue type, and terminate the old job thread.";
jobThread = null;
jobHandler = null;
}
// valid handler
if (jobHandler == null) {
jobHandler = newJobHandler;
if (jobHandler == null) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "job handler [" + triggerParam.getExecutorHandler() + "] not found.");
}
}
} else if (GlueTypeEnum.GLUE_GROOVY == glueTypeEnum) {
// valid old jobThread
if (jobThread != null &&
!(jobThread.getHandler() instanceof GlueJobHandler
&& ((GlueJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
// change handler or gluesource updated, need kill old thread
removeOldReason = "change job source or glue type, and terminate the old job thread.";
jobThread = null;
jobHandler = null;
}
// valid handler
if (jobHandler == null) {
try {
IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource());
jobHandler = new GlueJobHandler(originJobHandler, triggerParam.getGlueUpdatetime());
} catch (Exception e) {
logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, e.getMessage());
}
}
} else if (glueTypeEnum!=null && glueTypeEnum.isScript()) {
// valid old jobThread
if (jobThread != null &&
!(jobThread.getHandler() instanceof ScriptJobHandler
&& ((ScriptJobHandler) jobThread.getHandler()).getGlueUpdatetime()==triggerParam.getGlueUpdatetime() )) {
// change script or gluesource updated, need kill old thread
removeOldReason = "change job source or glue type, and terminate the old job thread.";
jobThread = null;
jobHandler = null;
}
// valid handler
if (jobHandler == null) {
jobHandler = new ScriptJobHandler(triggerParam.getJobId(), triggerParam.getGlueUpdatetime(), triggerParam.getGlueSource(), GlueTypeEnum.match(triggerParam.getGlueType()));
}
} else {
return new ReturnT<String>(ReturnT.FAIL_CODE, "glueType[" + triggerParam.getGlueType() + "] is not valid.");
}
// executor block strategy
if (jobThread != null) {
ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
// discard when running
if (jobThread.isRunningOrHasQueue()) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect"+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
}
} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
// kill running jobThread
if (jobThread.isRunningOrHasQueue()) {
removeOldReason = "block strategy effect" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();
jobThread = null;
}
} else {
// just queue trigger
}
}
// replace thread (new or exists invalid)
if (jobThread == null) {
jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
}
// push data to queue
ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);
return pushResult;
}
@Override
public ReturnT<String> kill(KillParam killParam) {
// kill handlerThread, and create new one
JobThread jobThread = XxlJobExecutor.loadJobThread(killParam.getJobId());
if (jobThread != null) {
XxlJobExecutor.removeJobThread(killParam.getJobId(), "scheduling center kill job.");
return ReturnT.SUCCESS;
}
return new ReturnT<String>(ReturnT.SUCCESS_CODE, "job thread already killed.");
}
@Override
public ReturnT<LogResult> log(LogParam logParam) {
// log filename: logPath/yyyy-MM-dd/9999.log
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(logParam.getLogDateTim()), logParam.getLogId());
LogResult logResult = XxlJobFileAppender.readLog(logFileName, logParam.getFromLineNum());
return new ReturnT<LogResult>(logResult);
}
}

View File

@@ -0,0 +1,68 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* Created by xuxueli on 17/3/2.
*/
public class HandleCallbackParam implements Serializable {
private static final long serialVersionUID = 42L;
private String logId;
private long logDateTim;
private int handleCode;
private String handleMsg;
public HandleCallbackParam(){}
public HandleCallbackParam(String logId, long logDateTim, int handleCode, String handleMsg) {
this.logId = logId;
this.logDateTim = logDateTim;
this.handleCode = handleCode;
this.handleMsg = handleMsg;
}
public String getLogId() {
return logId;
}
public void setLogId(String logId) {
this.logId = logId;
}
public long getLogDateTim() {
return logDateTim;
}
public void setLogDateTim(long logDateTim) {
this.logDateTim = logDateTim;
}
public int getHandleCode() {
return handleCode;
}
public void setHandleCode(int handleCode) {
this.handleCode = handleCode;
}
public String getHandleMsg() {
return handleMsg;
}
public void setHandleMsg(String handleMsg) {
this.handleMsg = handleMsg;
}
@Override
public String toString() {
return "HandleCallbackParam{" +
"logId=" + logId +
", logDateTim=" + logDateTim +
", handleCode=" + handleCode +
", handleMsg='" + handleMsg + '\'' +
'}';
}
}

View File

@@ -0,0 +1,29 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* @author xuxueli 2020-04-11 22:27
*/
public class IdleBeatParam implements Serializable {
private static final long serialVersionUID = 42L;
public IdleBeatParam() {
}
public IdleBeatParam(String jobId) {
this.jobId = jobId;
}
private String jobId;
public String getJobId() {
return jobId;
}
public void setJobId(String jobId) {
this.jobId = jobId;
}
}

View File

@@ -0,0 +1,29 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* @author xuxueli 2020-04-11 22:27
*/
public class KillParam implements Serializable {
private static final long serialVersionUID = 42L;
public KillParam() {
}
public KillParam(String jobId) {
this.jobId = jobId;
}
private String jobId;
public String getJobId() {
return jobId;
}
public void setJobId(String jobId) {
this.jobId = jobId;
}
}

View File

@@ -0,0 +1,48 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* @author xuxueli 2020-04-11 22:27
*/
public class LogParam implements Serializable {
private static final long serialVersionUID = 42L;
public LogParam() {
}
public LogParam(long logDateTim, String logId, int fromLineNum) {
this.logDateTim = logDateTim;
this.logId = logId;
this.fromLineNum = fromLineNum;
}
private long logDateTim;
private String logId;
private int fromLineNum;
public long getLogDateTim() {
return logDateTim;
}
public void setLogDateTim(long logDateTim) {
this.logDateTim = logDateTim;
}
public String getLogId() {
return logId;
}
public void setLogId(String logId) {
this.logId = logId;
}
public int getFromLineNum() {
return fromLineNum;
}
public void setFromLineNum(int fromLineNum) {
this.fromLineNum = fromLineNum;
}
}

View File

@@ -0,0 +1,57 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* Created by xuxueli on 17/3/23.
*/
public class LogResult implements Serializable {
private static final long serialVersionUID = 42L;
public LogResult() {
}
public LogResult(int fromLineNum, int toLineNum, String logContent, boolean isEnd) {
this.fromLineNum = fromLineNum;
this.toLineNum = toLineNum;
this.logContent = logContent;
this.isEnd = isEnd;
}
private int fromLineNum;
private int toLineNum;
private String logContent;
private boolean isEnd;
public int getFromLineNum() {
return fromLineNum;
}
public void setFromLineNum(int fromLineNum) {
this.fromLineNum = fromLineNum;
}
public int getToLineNum() {
return toLineNum;
}
public void setToLineNum(int toLineNum) {
this.toLineNum = toLineNum;
}
public String getLogContent() {
return logContent;
}
public void setLogContent(String logContent) {
this.logContent = logContent;
}
public boolean isEnd() {
return isEnd;
}
public void setEnd(boolean end) {
isEnd = end;
}
}

View File

@@ -0,0 +1,53 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
import java.util.List;
/**
* 注册handler参数
*
* @author :云筑产品开发平台组
* @version: V3.1.0
* @copyright 深圳市乐程软件有限公司
* @date 2022/3/29 11:36
*/
public class RegistryHandlerName implements Serializable {
/**
* handler集合
*/
private List<String> handlerList;
/**
* 执行器名称
*/
private String executorName;
/**
* 调用方法
*/
private String uri;
public List<String> getHandlerList() {
return handlerList;
}
public void setHandlerList(List<String> handlerList) {
this.handlerList = handlerList;
}
public String getExecutorName() {
return executorName;
}
public void setExecutorName(String executorName) {
this.executorName = executorName;
}
public String getUri() {
return uri;
}
public void setUri(String uri) {
this.uri = uri;
}
}

View File

@@ -0,0 +1,55 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* Created by xuxueli on 2017-05-10 20:22:42
*/
public class RegistryParam implements Serializable {
private static final long serialVersionUID = 42L;
private String registryGroup;
private String registryKey;
private String registryValue;
public RegistryParam(){}
public RegistryParam(String registryGroup, String registryKey, String registryValue) {
this.registryGroup = registryGroup;
this.registryKey = registryKey;
this.registryValue = registryValue;
}
public String getRegistryGroup() {
return registryGroup;
}
public void setRegistryGroup(String registryGroup) {
this.registryGroup = registryGroup;
}
public String getRegistryKey() {
return registryKey;
}
public void setRegistryKey(String registryKey) {
this.registryKey = registryKey;
}
public String getRegistryValue() {
return registryValue;
}
public void setRegistryValue(String registryValue) {
this.registryValue = registryValue;
}
@Override
public String toString() {
return "RegistryParam{" +
"registryGroup='" + registryGroup + '\'' +
", registryKey='" + registryKey + '\'' +
", registryValue='" + registryValue + '\'' +
'}';
}
}

View File

@@ -0,0 +1,58 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* common return
* @author xuxueli 2015-12-4 16:32:31
* @param <T>
*/
public class ReturnT<T> implements Serializable {
public static final long serialVersionUID = 42L;
public static final int SUCCESS_CODE = 200;
public static final int FAIL_CODE = 500;
public static final ReturnT<String> SUCCESS = new ReturnT<String>(null);
public static final ReturnT<String> FAIL = new ReturnT<String>(FAIL_CODE, null);
private int code;
private String msg;
private T content;
public ReturnT(){}
public ReturnT(int code, String msg) {
this.code = code;
this.msg = msg;
}
public ReturnT(T content) {
this.code = SUCCESS_CODE;
this.content = content;
}
public int getCode() {
return code;
}
public void setCode(int code) {
this.code = code;
}
public String getMsg() {
return msg;
}
public void setMsg(String msg) {
this.msg = msg;
}
public T getContent() {
return content;
}
public void setContent(T content) {
this.content = content;
}
@Override
public String toString() {
return "ReturnT [code=" + code + ", msg=" + msg + ", content=" + content + "]";
}
}

View File

@@ -0,0 +1,145 @@
package com.xxl.job.core.biz.model;
import java.io.Serializable;
/**
* Created by xuxueli on 16/7/22.
*/
public class TriggerParam implements Serializable{
private static final long serialVersionUID = 42L;
private String jobId;
private String executorHandler;
private String executorParams;
private String executorBlockStrategy;
private int executorTimeout;
private String logId;
private long logDateTime;
private String glueType;
private String glueSource;
private long glueUpdatetime;
private int broadcastIndex;
private int broadcastTotal;
public String getJobId() {
return jobId;
}
public void setJobId(String jobId) {
this.jobId = jobId;
}
public String getExecutorHandler() {
return executorHandler;
}
public void setExecutorHandler(String executorHandler) {
this.executorHandler = executorHandler;
}
public String getExecutorParams() {
return executorParams;
}
public void setExecutorParams(String executorParams) {
this.executorParams = executorParams;
}
public String getExecutorBlockStrategy() {
return executorBlockStrategy;
}
public void setExecutorBlockStrategy(String executorBlockStrategy) {
this.executorBlockStrategy = executorBlockStrategy;
}
public int getExecutorTimeout() {
return executorTimeout;
}
public void setExecutorTimeout(int executorTimeout) {
this.executorTimeout = executorTimeout;
}
public String getLogId() {
return logId;
}
public void setLogId(String logId) {
this.logId = logId;
}
public long getLogDateTime() {
return logDateTime;
}
public void setLogDateTime(long logDateTime) {
this.logDateTime = logDateTime;
}
public String getGlueType() {
return glueType;
}
public void setGlueType(String glueType) {
this.glueType = glueType;
}
public String getGlueSource() {
return glueSource;
}
public void setGlueSource(String glueSource) {
this.glueSource = glueSource;
}
public long getGlueUpdatetime() {
return glueUpdatetime;
}
public void setGlueUpdatetime(long glueUpdatetime) {
this.glueUpdatetime = glueUpdatetime;
}
public int getBroadcastIndex() {
return broadcastIndex;
}
public void setBroadcastIndex(int broadcastIndex) {
this.broadcastIndex = broadcastIndex;
}
public int getBroadcastTotal() {
return broadcastTotal;
}
public void setBroadcastTotal(int broadcastTotal) {
this.broadcastTotal = broadcastTotal;
}
@Override
public String toString() {
return "TriggerParam{" +
"jobId=" + jobId +
", executorHandler='" + executorHandler + '\'' +
", executorParams='" + executorParams + '\'' +
", executorBlockStrategy='" + executorBlockStrategy + '\'' +
", executorTimeout=" + executorTimeout +
", logId=" + logId +
", logDateTime=" + logDateTime +
", glueType='" + glueType + '\'' +
", glueSource='" + glueSource + '\'' +
", glueUpdatetime=" + glueUpdatetime +
", broadcastIndex=" + broadcastIndex +
", broadcastTotal=" + broadcastTotal +
'}';
}
}

View File

@@ -0,0 +1,123 @@
package com.xxl.job.core.context;
/**
* xxl-job context
*
* @author xuxueli 2020-05-21
* [Dear hj]
*/
public class XxlJobContext {
public static final int HANDLE_CODE_SUCCESS = 200;
public static final int HANDLE_CODE_FAIL = 500;
public static final int HANDLE_CODE_TIMEOUT = 502;
// ---------------------- base info ----------------------
/**
* job id
*/
private final String jobId;
/**
* job param
*/
private final String jobParam;
// ---------------------- for log ----------------------
/**
* job log filename
*/
private final String jobLogFileName;
// ---------------------- for shard ----------------------
/**
* shard index
*/
private final int shardIndex;
/**
* shard total
*/
private final int shardTotal;
// ---------------------- for handle ----------------------
/**
* handleCodeThe result status of job execution
*
* 200 : success
* 500 : fail
* 502 : timeout
*
*/
private int handleCode;
/**
* handleMsgThe simple log msg of job execution
*/
private String handleMsg;
public XxlJobContext(String jobId, String jobParam, String jobLogFileName, int shardIndex, int shardTotal) {
this.jobId = jobId;
this.jobParam = jobParam;
this.jobLogFileName = jobLogFileName;
this.shardIndex = shardIndex;
this.shardTotal = shardTotal;
this.handleCode = HANDLE_CODE_SUCCESS; // default success
}
public String getJobId() {
return jobId;
}
public String getJobParam() {
return jobParam;
}
public String getJobLogFileName() {
return jobLogFileName;
}
public int getShardIndex() {
return shardIndex;
}
public int getShardTotal() {
return shardTotal;
}
public void setHandleCode(int handleCode) {
this.handleCode = handleCode;
}
public int getHandleCode() {
return handleCode;
}
public void setHandleMsg(String handleMsg) {
this.handleMsg = handleMsg;
}
public String getHandleMsg() {
return handleMsg;
}
// ---------------------- tool ----------------------
private static InheritableThreadLocal<XxlJobContext> contextHolder = new InheritableThreadLocal<XxlJobContext>(); // support for child thread of job handler)
public static void setXxlJobContext(XxlJobContext xxlJobContext){
contextHolder.set(xxlJobContext);
}
public static XxlJobContext getXxlJobContext(){
return contextHolder.get();
}
}

View File

@@ -0,0 +1,256 @@
package com.xxl.job.core.context;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.util.DateUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.helpers.FormattingTuple;
import org.slf4j.helpers.MessageFormatter;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Date;
/**
* helper for xxl-job
*
* @author xuxueli 2020-11-05
*/
public class XxlJobHelper {
// ---------------------- base info ----------------------
/**
* current JobId
*
* @return
*/
public static String getJobId() {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return "0";
}
return xxlJobContext.getJobId();
}
/**
* current JobParam
*
* @return
*/
public static String getJobParam() {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return null;
}
return xxlJobContext.getJobParam();
}
// ---------------------- for log ----------------------
/**
* current JobLogFileName
*
* @return
*/
public static String getJobLogFileName() {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return null;
}
return xxlJobContext.getJobLogFileName();
}
// ---------------------- for shard ----------------------
/**
* current ShardIndex
*
* @return
*/
public static int getShardIndex() {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return -1;
}
return xxlJobContext.getShardIndex();
}
/**
* current ShardTotal
*
* @return
*/
public static int getShardTotal() {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return -1;
}
return xxlJobContext.getShardTotal();
}
// ---------------------- tool for log ----------------------
private static Logger logger = LoggerFactory.getLogger("xxl-job logger");
/**
* append log with pattern
*
* @param appendLogPattern like "aaa {} bbb {} ccc"
* @param appendLogArguments like "111, true"
*/
public static boolean log(String appendLogPattern, Object ... appendLogArguments) {
FormattingTuple ft = MessageFormatter.arrayFormat(appendLogPattern, appendLogArguments);
String appendLog = ft.getMessage();
/*appendLog = appendLogPattern;
if (appendLogArguments!=null && appendLogArguments.length>0) {
appendLog = MessageFormat.format(appendLogPattern, appendLogArguments);
}*/
StackTraceElement callInfo = new Throwable().getStackTrace()[1];
return logDetail(callInfo, appendLog);
}
/**
* append exception stack
*
* @param e
*/
public static boolean log(Throwable e) {
StringWriter stringWriter = new StringWriter();
e.printStackTrace(new PrintWriter(stringWriter));
String appendLog = stringWriter.toString();
StackTraceElement callInfo = new Throwable().getStackTrace()[1];
return logDetail(callInfo, appendLog);
}
/**
* append log
*
* @param callInfo
* @param appendLog
*/
private static boolean logDetail(StackTraceElement callInfo, String appendLog) {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return false;
}
/*// "yyyy-MM-dd HH:mm:ss [ClassName]-[MethodName]-[LineNumber]-[ThreadName] log";
StackTraceElement[] stackTraceElements = new Throwable().getStackTrace();
StackTraceElement callInfo = stackTraceElements[1];*/
StringBuffer stringBuffer = new StringBuffer();
stringBuffer.append(DateUtil.formatDateTime(new Date())).append(" ")
.append("["+ callInfo.getClassName() + "#" + callInfo.getMethodName() +"]").append("-")
.append("["+ callInfo.getLineNumber() +"]").append("-")
.append("["+ Thread.currentThread().getName() +"]").append(" ")
.append(appendLog!=null?appendLog:"");
String formatAppendLog = stringBuffer.toString();
// appendlog
String logFileName = xxlJobContext.getJobLogFileName();
if (logFileName!=null && logFileName.trim().length()>0) {
XxlJobFileAppender.appendLog(logFileName, formatAppendLog);
return true;
} else {
logger.info(">>>>>>>>>>> {}", formatAppendLog);
return false;
}
}
// ---------------------- tool for handleResult ----------------------
/**
* handle success
*
* @return
*/
public static boolean handleSuccess(){
return handleResult(XxlJobContext.HANDLE_CODE_SUCCESS, null);
}
/**
* handle success with log msg
*
* @param handleMsg
* @return
*/
public static boolean handleSuccess(String handleMsg) {
return handleResult(XxlJobContext.HANDLE_CODE_SUCCESS, handleMsg);
}
/**
* handle fail
*
* @return
*/
public static boolean handleFail(){
return handleResult(XxlJobContext.HANDLE_CODE_FAIL, null);
}
/**
* handle fail with log msg
*
* @param handleMsg
* @return
*/
public static boolean handleFail(String handleMsg) {
return handleResult(XxlJobContext.HANDLE_CODE_FAIL, handleMsg);
}
/**
* handle timeout
*
* @return
*/
public static boolean handleTimeout(){
return handleResult(XxlJobContext.HANDLE_CODE_TIMEOUT, null);
}
/**
* handle timeout with log msg
*
* @param handleMsg
* @return
*/
public static boolean handleTimeout(String handleMsg){
return handleResult(XxlJobContext.HANDLE_CODE_TIMEOUT, handleMsg);
}
/**
* @param handleCode
*
* 200 : success
* 500 : fail
* 502 : timeout
*
* @param handleMsg
* @return
*/
public static boolean handleResult(int handleCode, String handleMsg) {
XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
if (xxlJobContext == null) {
return false;
}
xxlJobContext.setHandleCode(handleCode);
if (handleMsg != null) {
xxlJobContext.setHandleMsg(handleMsg);
}
return true;
}
}

View File

@@ -0,0 +1,36 @@
package com.xxl.job.core.enums;
/**
* Created by xuxueli on 17/5/9.
*/
public enum ExecutorBlockStrategyEnum {
SERIAL_EXECUTION("Serial execution"),
/*CONCURRENT_EXECUTION("并行"),*/
DISCARD_LATER("Discard Later"),
COVER_EARLY("Cover Early");
private String title;
private ExecutorBlockStrategyEnum (String title) {
this.title = title;
}
public void setTitle(String title) {
this.title = title;
}
public String getTitle() {
return title;
}
public static ExecutorBlockStrategyEnum match(String name, ExecutorBlockStrategyEnum defaultItem) {
if (name != null) {
for (ExecutorBlockStrategyEnum item:ExecutorBlockStrategyEnum.values()) {
if (item.name().equals(name)) {
return item;
}
}
}
return defaultItem;
}
}

View File

@@ -0,0 +1,14 @@
package com.xxl.job.core.enums;
/**
* Created by xuxueli on 17/5/10.
*/
public class RegistryConfig {
public static final int BEAT_TIMEOUT = 30;
public static final int DEAD_TIMEOUT = BEAT_TIMEOUT * 3;
public enum RegistType{ EXECUTOR, ADMIN }
}

View File

@@ -0,0 +1,337 @@
package com.xxl.job.core.executor;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.client.AdminBizClient;
import com.xxl.job.core.biz.model.RegistryHandlerName;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.handler.annotation.XxlJob;
import com.xxl.job.core.handler.impl.MethodJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.server.EmbedServer;
import com.xxl.job.core.thread.JobLogFileCleanThread;
import com.xxl.job.core.thread.JobThread;
import com.xxl.job.core.thread.TriggerCallbackThread;
import com.xxl.job.core.util.IpUtil;
import com.xxl.job.core.util.NetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* Created by xuxueli on 2016/3/2 21:14.
*/
public class XxlJobExecutor {
private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
// ---------------------- param ----------------------
private String adminAddresses;
private String accessToken;
private String appname;
private String address;
private String ip;
private int port;
private String logPath;
private int logRetentionDays;
public void setAdminAddresses(String adminAddresses) {
this.adminAddresses = adminAddresses;
}
public void setAccessToken(String accessToken) {
this.accessToken = accessToken;
}
public void setAppname(String appname) {
this.appname = appname;
}
public void setAddress(String address) {
this.address = address;
}
public void setIp(String ip) {
this.ip = ip;
}
public void setPort(int port) {
this.port = port;
}
public void setLogPath(String logPath) {
this.logPath = logPath;
}
public void setLogRetentionDays(int logRetentionDays) {
this.logRetentionDays = logRetentionDays;
}
// ---------------------- start + stop ----------------------
public void start() throws Exception {
// init logpath
XxlJobFileAppender.initLogPath(logPath);
// init invoker, admin-client
initAdminBizList(adminAddresses, accessToken);
// init JobLogFileCleanThread
JobLogFileCleanThread.getInstance().start(logRetentionDays);
// init TriggerCallbackThread
TriggerCallbackThread.getInstance().start();
// init executor-server
initEmbedServer(address, ip, port, appname, accessToken);
}
public void destroy(){
// for (AdminBiz adminBiz : XxlJobExecutor.getAdminBizList()) {
// try {
// // 构建模型
// RegistryHandlerName registryHandlerName = new RegistryHandlerName();
// registryHandlerName.setExecutorName(appname);
// // 存放handlerName
// List<String> list = new ArrayList<>(16);
// for (String key : jobHandlerRepository.keySet()) {
// if ("defaultHandler".equals(key)) {
// continue;
// }
// list.add(key);
// }
// // 走注册
// registryHandlerName.setUri("remove");
// registryHandlerName.setHandlerList(list);
// ReturnT<String> registryResult = registryHandlerName.getHandlerList().size() > 0 ? adminBiz.registryHandlerName(registryHandlerName) : new ReturnT<String>("注册成功");
// if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
// registryResult = ReturnT.SUCCESS;
// logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{jobHandlerRepository, registryResult});
// break;
// } else {
// logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{jobHandlerRepository, registryResult});
// }
// } catch (Exception e) {
//// if (!toStop) {
// logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", jobHandlerRepository, e);
//// }
//
// }
// }
// destroy executor-server
stopEmbedServer();
// destroy jobThreadRepository
if (jobThreadRepository.size() > 0) {
for (Map.Entry<String, JobThread> item: jobThreadRepository.entrySet()) {
JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
// wait for job thread push result to callback queue
if (oldJobThread != null) {
try {
oldJobThread.join();
} catch (InterruptedException e) {
logger.error(">>>>>>>>>>> xxl-job, JobThread destroy(join) error, jobId:{}", item.getKey(), e);
}
}
}
jobThreadRepository.clear();
}
jobHandlerRepository.clear();
// destroy JobLogFileCleanThread
JobLogFileCleanThread.getInstance().toStop();
// destroy TriggerCallbackThread
TriggerCallbackThread.getInstance().toStop();
}
// ---------------------- admin-client (rpc invoker) ----------------------
private static List<AdminBiz> adminBizList;
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
if (adminAddresses!=null && adminAddresses.trim().length()>0) {
for (String address: adminAddresses.trim().split(",")) {
if (address!=null && address.trim().length()>0) {
AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
if (adminBizList == null) {
adminBizList = new ArrayList<AdminBiz>();
}
adminBizList.add(adminBiz);
}
}
}
}
public static List<AdminBiz> getAdminBizList(){
return adminBizList;
}
// ---------------------- executor-server (rpc provider) ----------------------
private EmbedServer embedServer = null;
private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
for (AdminBiz adminBiz : XxlJobExecutor.getAdminBizList()) {
try {
// 构建模型
RegistryHandlerName registryHandlerName = new RegistryHandlerName();
registryHandlerName.setExecutorName(appname);
// 存放handlerName
List<String> list = new ArrayList<>(16);
for (String key : jobHandlerRepository.keySet()) {
if ("defaultHandler".equals(key)) {
continue;
}
list.add(key);
}
// 走注册
registryHandlerName.setUri("register");
registryHandlerName.setHandlerList(list);
ReturnT<String> registryResult = registryHandlerName.getHandlerList().size() > 0 ? adminBiz.registryHandlerName(registryHandlerName) : new ReturnT<String>("注册成功");
if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
registryResult = ReturnT.SUCCESS;
logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{jobHandlerRepository, registryResult});
break;
} else {
logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{jobHandlerRepository, registryResult});
}
} catch (Exception e) {
// if (!toStop) {
logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", jobHandlerRepository, e);
// }
}
}
// fill ip port
port = port>0?port: NetUtil.findAvailablePort(9999);
ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
// generate address
if (address==null || address.trim().length()==0) {
String ip_port_address = IpUtil.getIpPort(ip, port); // registry-addressdefault use address to registry , otherwise use ip:port if address is null
address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
}
// accessToken
if (accessToken==null || accessToken.trim().length()==0) {
logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
}
// start
embedServer = new EmbedServer();
embedServer.start(address, port, appname, accessToken);
}
private void stopEmbedServer() {
// stop provider factory
if (embedServer != null) {
try {
embedServer.stop();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
// ---------------------- job handler repository ----------------------
private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
public static IJobHandler loadJobHandler(String name){
return jobHandlerRepository.get(name);
}
public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
return jobHandlerRepository.put(name, jobHandler);
}
protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod){
if (xxlJob == null) {
return;
}
String name = xxlJob.value();
//make and simplify the variables since they'll be called several times later
Class<?> clazz = bean.getClass();
String methodName = executeMethod.getName();
if (name.trim().length() == 0) {
throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + clazz + "#" + methodName + "] .");
}
if (loadJobHandler(name) != null) {
throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
}
// execute method
/*if (!(method.getParameterTypes().length == 1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) {
throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
"The correct method format like \" public ReturnT<String> execute(String param) \" .");
}
if (!method.getReturnType().isAssignableFrom(ReturnT.class)) {
throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
"The correct method format like \" public ReturnT<String> execute(String param) \" .");
}*/
executeMethod.setAccessible(true);
// init and destroy
Method initMethod = null;
Method destroyMethod = null;
if (xxlJob.init().trim().length() > 0) {
try {
initMethod = clazz.getDeclaredMethod(xxlJob.init());
initMethod.setAccessible(true);
} catch (NoSuchMethodException e) {
throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + clazz + "#" + methodName + "] .");
}
}
if (xxlJob.destroy().trim().length() > 0) {
try {
destroyMethod = clazz.getDeclaredMethod(xxlJob.destroy());
destroyMethod.setAccessible(true);
} catch (NoSuchMethodException e) {
throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + clazz + "#" + methodName + "] .");
}
}
// registry jobhandler
registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));
}
// ---------------------- job thread repository ----------------------
private static ConcurrentMap<String, JobThread> jobThreadRepository = new ConcurrentHashMap<>();
public static JobThread registJobThread(String jobId, IJobHandler handler, String removeOldReason){
JobThread newJobThread = new JobThread(jobId, handler);
newJobThread.start();
logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});
JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); // putIfAbsent | oh my god, map's put method return the old value!!!
if (oldJobThread != null) {
oldJobThread.toStop(removeOldReason);
oldJobThread.interrupt();
}
return newJobThread;
}
public static JobThread removeJobThread(String jobId, String removeOldReason){
JobThread oldJobThread = jobThreadRepository.remove(jobId);
if (oldJobThread != null) {
oldJobThread.toStop(removeOldReason);
oldJobThread.interrupt();
return oldJobThread;
}
return null;
}
public static JobThread loadJobThread(String jobId){
return jobThreadRepository.get(jobId);
}
}

View File

@@ -0,0 +1,68 @@
package com.xxl.job.core.executor.impl;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.handler.annotation.XxlJob;
import java.lang.reflect.Method;
import java.util.ArrayList;
import java.util.List;
/**
* xxl-job executor (for frameless)
*
* @author xuxueli 2020-11-05
*/
public class XxlJobSimpleExecutor extends XxlJobExecutor {
private List<Object> xxlJobBeanList = new ArrayList<>();
public List<Object> getXxlJobBeanList() {
return xxlJobBeanList;
}
public void setXxlJobBeanList(List<Object> xxlJobBeanList) {
this.xxlJobBeanList = xxlJobBeanList;
}
@Override
public void start() {
// init JobHandler Repository (for method)
initJobHandlerMethodRepository(xxlJobBeanList);
// super start
try {
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
@Override
public void destroy() {
super.destroy();
}
private void initJobHandlerMethodRepository(List<Object> xxlJobBeanList) {
if (xxlJobBeanList == null || xxlJobBeanList.size() == 0) {
return;
}
// init job handler from method
for (Object bean : xxlJobBeanList) {
// method
Method[] methods = bean.getClass().getDeclaredMethods();
if (methods.length == 0) {
continue;
}
for (Method executeMethod : methods) {
XxlJob xxlJob = executeMethod.getAnnotation(XxlJob.class);
// registry
registJobHandler(xxlJob, bean, executeMethod);
}
}
}
}

View File

@@ -0,0 +1,155 @@
package com.xxl.job.core.executor.impl;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.glue.GlueFactory;
import com.xxl.job.core.handler.annotation.XxlJob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.SmartInitializingSingleton;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.MethodIntrospector;
import org.springframework.core.annotation.AnnotatedElementUtils;
import org.springframework.lang.NonNull;
import java.lang.reflect.Method;
import java.util.Map;
/**
* xxl-job executor (for spring)
*
* @author xuxueli 2018-11-01 09:24:52
*/
public class XxlJobSpringExecutor extends XxlJobExecutor
implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean {
private static final Logger logger = LoggerFactory.getLogger(XxlJobSpringExecutor.class);
// start
@Override
public void afterSingletonsInstantiated() {
// init JobHandler Repository
/* initJobHandlerRepository(applicationContext); */
// init JobHandler Repository (for method)
initJobHandlerMethodRepository(applicationContext);
// refresh GlueFactory
GlueFactory.refreshInstance(1);
// super start
try {
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
// destroy
@Override
public void destroy() {
super.destroy();
}
/*
* private void initJobHandlerRepository(ApplicationContext applicationContext)
* {
* if (applicationContext == null) {
* return;
* }
*
* // init job handler action
* Map<String, Object> serviceBeanMap =
* applicationContext.getBeansWithAnnotation(JobHandler.class);
*
* if (serviceBeanMap != null && serviceBeanMap.size() > 0) {
* for (Object serviceBean : serviceBeanMap.values()) {
* if (serviceBean instanceof IJobHandler) {
* String name = serviceBean.getClass().getAnnotation(JobHandler.class).value();
* IJobHandler handler = (IJobHandler) serviceBean;
* if (loadJobHandler(name) != null) {
* throw new RuntimeException("xxl-job jobhandler[" + name +
* "] naming conflicts.");
* }
* registJobHandler(name, handler);
* }
* }
* }
* }
*/
@SuppressWarnings("null")
private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
if (applicationContext == null) {
return;
}
// init job handler from method
String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
for (String beanDefinitionName : beanDefinitionNames) {
// get bean
Object bean = null;
Lazy onBean = applicationContext.findAnnotationOnBean(beanDefinitionName, Lazy.class);
if (onBean != null) {
logger.debug("xxl-job annotation scan, skip @Lazy Bean:{}", beanDefinitionName);
continue;
} else {
bean = applicationContext.getBean(beanDefinitionName);
}
// filter method
Map<Method, XxlJob> annotatedMethods = null; // referred to
// org.springframework.context.event.EventListenerMethodProcessor.processBean
try {
annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),
new MethodIntrospector.MetadataLookup<XxlJob>() {
@Override
public XxlJob inspect(@NonNull Method method) {
return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class);
}
});
} catch (Throwable ex) {
logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex);
}
if (annotatedMethods == null || annotatedMethods.isEmpty()) {
continue;
}
// generate and regist method job handler
for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {
Method executeMethod = methodXxlJobEntry.getKey();
XxlJob xxlJob = methodXxlJobEntry.getValue();
// regist
registJobHandler(xxlJob, bean, executeMethod);
}
}
}
// ---------------------- applicationContext ----------------------
private static ApplicationContext applicationContext;
@Override
public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
XxlJobSpringExecutor.applicationContext = applicationContext;
}
public static ApplicationContext getApplicationContext() {
return applicationContext;
}
/*
* BeanDefinitionRegistryPostProcessor
* registry.getBeanDefine()
*
* @Override
* public void postProcessBeanDefinitionRegistry(BeanDefinitionRegistry
* registry) throws BeansException {
* this.registry = registry;
* }
*/
}

View File

@@ -0,0 +1,91 @@
package com.xxl.job.core.glue;
import com.xxl.job.core.glue.impl.SpringGlueFactory;
import com.xxl.job.core.handler.IJobHandler;
import groovy.lang.GroovyClassLoader;
import java.math.BigInteger;
import java.security.MessageDigest;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
/**
* glue factory, product class/object by name
*
* @author xuxueli 2016-1-2 20:02:27
*/
public class GlueFactory {
private static GlueFactory glueFactory = new GlueFactory();
public static GlueFactory getInstance() {
return glueFactory;
}
public static void refreshInstance(int type) {
if (type == 0) {
glueFactory = new GlueFactory();
} else if (type == 1) {
glueFactory = new SpringGlueFactory();
}
}
/**
* groovy class loader
*/
private GroovyClassLoader groovyClassLoader = new GroovyClassLoader();
private ConcurrentMap<String, Class<?>> CLASS_CACHE = new ConcurrentHashMap<>();
/**
* load new instance, prototype
*
* @param codeSource
* @return
* @throws Exception
*/
public IJobHandler loadNewInstance(String codeSource) throws Exception {
if (codeSource != null && codeSource.trim().length() > 0) {
Class<?> clazz = getCodeSourceClass(codeSource);
if (clazz != null) {
Object instance = clazz.getDeclaredConstructor().newInstance();
if (instance != null) {
if (instance instanceof IJobHandler) {
this.injectService(instance);
return (IJobHandler) instance;
} else {
throw new IllegalArgumentException(">>>>>>>>>>> xxl-glue, loadNewInstance error, "
+ "cannot convert from instance[" + instance.getClass() + "] to IJobHandler");
}
}
}
}
throw new IllegalArgumentException(">>>>>>>>>>> xxl-glue, loadNewInstance error, instance is null");
}
private Class<?> getCodeSourceClass(String codeSource) {
try {
// md5
byte[] md5 = MessageDigest.getInstance("MD5").digest(codeSource.getBytes());
String md5Str = new BigInteger(1, md5).toString(16);
Class<?> clazz = CLASS_CACHE.get(md5Str);
if (clazz == null) {
clazz = groovyClassLoader.parseClass(codeSource);
CLASS_CACHE.putIfAbsent(md5Str, clazz);
}
return clazz;
} catch (Exception e) {
return groovyClassLoader.parseClass(codeSource);
}
}
/**
* inject service of bean field
*
* @param instance
*/
public void injectService(Object instance) {
// do something
}
}

View File

@@ -0,0 +1,54 @@
package com.xxl.job.core.glue;
/**
* Created by xuxueli on 17/4/26.
*/
public enum GlueTypeEnum {
BEAN("BEAN", false, null, null),
GLUE_GROOVY("GLUE(Java)", false, null, null),
GLUE_SHELL("GLUE(Shell)", true, "bash", ".sh"),
GLUE_PYTHON("GLUE(Python)", true, "python", ".py"),
GLUE_PHP("GLUE(PHP)", true, "php", ".php"),
GLUE_NODEJS("GLUE(Nodejs)", true, "node", ".js"),
GLUE_POWERSHELL("GLUE(PowerShell)", true, "powershell", ".ps1");
private String desc;
private boolean isScript;
private String cmd;
private String suffix;
private GlueTypeEnum(String desc, boolean isScript, String cmd, String suffix) {
this.desc = desc;
this.isScript = isScript;
this.cmd = cmd;
this.suffix = suffix;
}
public String getDesc() {
return desc;
}
public boolean isScript() {
return isScript;
}
public String getCmd() {
return cmd;
}
public String getSuffix() {
return suffix;
}
public static GlueTypeEnum match(String name){
for (GlueTypeEnum item: GlueTypeEnum.values()) {
if (item.name().equals(name)) {
return item;
}
}
return null;
}
}

View File

@@ -0,0 +1,82 @@
package com.xxl.job.core.glue.impl;
import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
import com.xxl.job.core.glue.GlueFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.core.annotation.AnnotationUtils;
import jakarta.annotation.Resource;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
/**
* @author xuxueli 2018-11-01
*/
public class SpringGlueFactory extends GlueFactory {
private static Logger logger = LoggerFactory.getLogger(SpringGlueFactory.class);
/**
* inject action of spring
*
* @param instance
*/
@Override
@SuppressWarnings("null")
public void injectService(Object instance) {
if (instance == null) {
return;
}
if (XxlJobSpringExecutor.getApplicationContext() == null) {
return;
}
Field[] fields = instance.getClass().getDeclaredFields();
for (Field field : fields) {
if (Modifier.isStatic(field.getModifiers())) {
continue;
}
Object fieldBean = null;
// with bean-id, bean could be found by both @Resource and @Autowired, or bean
// could only be found by @Autowired
Resource resourceAnnotation = AnnotationUtils.getAnnotation(field, Resource.class);
if (resourceAnnotation != null) {
try {
if (resourceAnnotation.name() != null && resourceAnnotation.name().length() > 0) {
fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(resourceAnnotation.name());
} else {
fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(field.getName());
}
} catch (Exception e) {
}
if (fieldBean == null) {
fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(field.getType());
}
} else if (AnnotationUtils.getAnnotation(field, Autowired.class) != null) {
Qualifier qualifier = AnnotationUtils.getAnnotation(field, Qualifier.class);
if (qualifier != null && qualifier.value() != null && qualifier.value().length() > 0) {
fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(qualifier.value());
} else {
fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(field.getType());
}
}
if (fieldBean != null) {
field.setAccessible(true);
try {
field.set(instance, fieldBean);
} catch (IllegalArgumentException e) {
logger.error(e.getMessage(), e);
} catch (IllegalAccessException e) {
logger.error(e.getMessage(), e);
}
}
}
}
}

View File

@@ -0,0 +1,39 @@
package com.xxl.job.core.handler;
/**
* job handler
*
* @author xuxueli 2015-12-19 19:06:38
*/
public abstract class IJobHandler {
/**
* execute handler, invoked when executor receives a scheduling request
*
* @throws Exception
*/
public abstract void execute() throws Exception;
/*@Deprecated
public abstract ReturnT<String> execute(String param) throws Exception;*/
/**
* init handler, invoked when JobThread init
*/
public void init() throws Exception {
// do something
}
/**
* destroy handler, invoked when JobThread destroy
*/
public void destroy() throws Exception {
// do something
}
}

View File

@@ -0,0 +1,49 @@
/**
* yunzhupaas Commercial License Agreement
* Copyright (c) 2018-2099, http://www.szlecheng.cn. All rights reserved.
* <p>
* Use of this software is governed by the Commercial License Agreement
* obtained after purchasing a license from yunzhupaas.
* <p>
* 1. This software is for development use only under a valid license
* from yunzhupaas.
* <p>
* 2. Redistribution of this software's source code to any third party
* without a commercial license is strictly prohibited.
* <p>
* 3. Licensees may copyright their own code but cannot use segments
* from this software for such purposes. Copyright of this software
* remains with yunzhupaas.
* <p>
* Using this software signifies agreement to this License, and the software
* must not be used for illegal purposes.
* <p>
* THIS yunzhupaasSoftWare IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
* not liable for any claims arising from secondary or illegal development.
* <p>
* Author: yunzhupaas.com
*/
// package com.xxl.job.core.handler.annotation;
//
// import java.lang.annotation.ElementType;
// import java.lang.annotation.Inherited;
// import java.lang.annotation.Retention;
// import java.lang.annotation.RetentionPolicy;
// import java.lang.annotation.Target;
//
/// **
// * annotation for job handler
// *
// * will be replaced by {@link com.xxl.job.core.handler.annotation.XxlJob}
// *
// * @author 2016-5-17 21:06:49
// */
// @Target({ElementType.TYPE})
// @Retention(RetentionPolicy.RUNTIME)
// @Inherited
// @Deprecated
// public @interface JobHandler {
//
// String value();
//
// }

View File

@@ -0,0 +1,31 @@
package com.xxl.job.core.handler.annotation;
import java.lang.annotation.*;
/**
* annotation for method jobhandler
*
* @author xuxueli 2019-12-11 20:50:13
*/
@Target({ElementType.METHOD})
@Retention(RetentionPolicy.RUNTIME)
@Inherited
public @interface XxlJob {
/**
* jobhandler name
*/
String value();
/**
* init handler, invoked when JobThread init
*/
String init() default "";
/**
* destroy handler, invoked when JobThread destroy
*/
String destroy() default "";
}

View File

@@ -0,0 +1,39 @@
package com.xxl.job.core.handler.impl;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.handler.IJobHandler;
/**
* glue job handler
*
* @author xuxueli 2016-5-19 21:05:45
*/
public class GlueJobHandler extends IJobHandler {
private long glueUpdatetime;
private IJobHandler jobHandler;
public GlueJobHandler(IJobHandler jobHandler, long glueUpdatetime) {
this.jobHandler = jobHandler;
this.glueUpdatetime = glueUpdatetime;
}
public long getGlueUpdatetime() {
return glueUpdatetime;
}
@Override
public void execute() throws Exception {
XxlJobHelper.log("----------- glue.version:"+ glueUpdatetime +" -----------");
jobHandler.execute();
}
@Override
public void init() throws Exception {
this.jobHandler.init();
}
@Override
public void destroy() throws Exception {
this.jobHandler.destroy();
}
}

View File

@@ -0,0 +1,54 @@
package com.xxl.job.core.handler.impl;
import com.xxl.job.core.handler.IJobHandler;
import java.lang.reflect.Method;
/**
* @author xuxueli 2019-12-11 21:12:18
*/
public class MethodJobHandler extends IJobHandler {
private final Object target;
private final Method method;
private Method initMethod;
private Method destroyMethod;
public MethodJobHandler(Object target, Method method, Method initMethod, Method destroyMethod) {
this.target = target;
this.method = method;
this.initMethod = initMethod;
this.destroyMethod = destroyMethod;
}
@Override
public void execute() throws Exception {
Class<?>[] paramTypes = method.getParameterTypes();
if (paramTypes.length > 0) {
method.invoke(target, new Object[paramTypes.length]); // method-param can not be primitive-types
} else {
method.invoke(target);
}
}
@Override
public void init() throws Exception {
if(initMethod != null) {
initMethod.invoke(target);
}
}
@Override
public void destroy() throws Exception {
if(destroyMethod != null) {
destroyMethod.invoke(target);
}
}
@Override
public String toString() {
return super.toString()+"["+ target.getClass() + "#" + method.getName() +"]";
}
}

View File

@@ -0,0 +1,94 @@
package com.xxl.job.core.handler.impl;
import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.glue.GlueTypeEnum;
import com.xxl.job.core.handler.IJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.util.ScriptUtil;
import java.io.File;
/**
* Created by xuxueli on 17/4/27.
*/
public class ScriptJobHandler extends IJobHandler {
private String jobId;
private long glueUpdatetime;
private String gluesource;
private GlueTypeEnum glueType;
public ScriptJobHandler(String jobId, long glueUpdatetime, String gluesource, GlueTypeEnum glueType){
this.jobId = jobId;
this.glueUpdatetime = glueUpdatetime;
this.gluesource = gluesource;
this.glueType = glueType;
// clean old script file
File glueSrcPath = new File(XxlJobFileAppender.getGlueSrcPath());
if (glueSrcPath.exists()) {
File[] glueSrcFileList = glueSrcPath.listFiles();
if (glueSrcFileList!=null && glueSrcFileList.length>0) {
for (File glueSrcFileItem : glueSrcFileList) {
if (glueSrcFileItem.getName().startsWith(String.valueOf(jobId)+"_")) {
glueSrcFileItem.delete();
}
}
}
}
}
public long getGlueUpdatetime() {
return glueUpdatetime;
}
@Override
public void execute() throws Exception {
if (!glueType.isScript()) {
XxlJobHelper.handleFail("glueType["+ glueType +"] invalid.");
return;
}
// cmd
String cmd = glueType.getCmd();
// make script file
String scriptFileName = XxlJobFileAppender.getGlueSrcPath()
.concat(File.separator)
.concat(String.valueOf(jobId))
.concat("_")
.concat(String.valueOf(glueUpdatetime))
.concat(glueType.getSuffix());
File scriptFile = new File(scriptFileName);
if (!scriptFile.exists()) {
ScriptUtil.markScriptFile(scriptFileName, gluesource);
}
// log file
String logFileName = XxlJobContext.getXxlJobContext().getJobLogFileName();
// script params0=param、1=分片序号、2=分片总数
String[] scriptParams = new String[3];
scriptParams[0] = XxlJobHelper.getJobParam();
scriptParams[1] = String.valueOf(XxlJobContext.getXxlJobContext().getShardIndex());
scriptParams[2] = String.valueOf(XxlJobContext.getXxlJobContext().getShardTotal());
// invoke
XxlJobHelper.log("----------- script file:"+ scriptFileName +" -----------");
int exitValue = ScriptUtil.execToFile(cmd, scriptFileName, logFileName, scriptParams);
if (exitValue == 0) {
XxlJobHelper.handleSuccess();
return;
} else {
XxlJobHelper.handleFail("script exit value("+exitValue+") is failed");
return ;
}
}
}

View File

@@ -0,0 +1,221 @@
package com.xxl.job.core.log;
import com.xxl.job.core.biz.model.LogResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
import java.text.SimpleDateFormat;
import java.util.Date;
/**
* store trigger log in each log-file
* @author xuxueli 2016-3-12 19:25:12
*/
public class XxlJobFileAppender {
private static Logger logger = LoggerFactory.getLogger(XxlJobFileAppender.class);
/**
* log base path
*
* strut like:
* ---/
* ---/gluesource/
* ---/gluesource/10_1514171108000.js
* ---/gluesource/10_1514171108000.js
* ---/2017-12-25/
* ---/2017-12-25/639.log
* ---/2017-12-25/821.log
*
*/
private static String logBasePath = "/data/applogs/xxl-job/jobhandler";
private static String glueSrcPath = logBasePath.concat("/gluesource");
public static void initLogPath(String logPath){
// init
if (logPath!=null && logPath.trim().length()>0) {
logBasePath = logPath;
}
// mk base dir
File logPathDir = new File(logBasePath);
if (!logPathDir.exists()) {
logPathDir.mkdirs();
}
logBasePath = logPathDir.getPath();
// mk glue dir
File glueBaseDir = new File(logPathDir, "gluesource");
if (!glueBaseDir.exists()) {
glueBaseDir.mkdirs();
}
glueSrcPath = glueBaseDir.getPath();
}
public static String getLogPath() {
return logBasePath;
}
public static String getGlueSrcPath() {
return glueSrcPath;
}
/**
* log filename, like "logPath/yyyy-MM-dd/9999.log"
*
* @param triggerDate
* @param logId
* @return
*/
public static String makeLogFileName(Date triggerDate, String logId) {
// filePath/yyyy-MM-dd
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd"); // avoid concurrent problem, can not be static
File logFilePath = new File(getLogPath(), sdf.format(triggerDate));
if (!logFilePath.exists()) {
logFilePath.mkdir();
}
// filePath/yyyy-MM-dd/9999.log
String logFileName = logFilePath.getPath()
.concat(File.separator)
.concat(logId)
.concat(".log");
return logFileName;
}
/**
* append log
*
* @param logFileName
* @param appendLog
*/
public static void appendLog(String logFileName, String appendLog) {
// log file
if (logFileName==null || logFileName.trim().length()==0) {
return;
}
File logFile = new File(logFileName);
if (!logFile.exists()) {
try {
logFile.createNewFile();
} catch (IOException e) {
logger.error(e.getMessage(), e);
return;
}
}
// log
if (appendLog == null) {
appendLog = "";
}
appendLog += "\r\n";
// append file content
FileOutputStream fos = null;
try {
fos = new FileOutputStream(logFile, true);
fos.write(appendLog.getBytes("utf-8"));
fos.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (fos != null) {
try {
fos.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
/**
* support read log-file
*
* @param logFileName
* @return log content
*/
public static LogResult readLog(String logFileName, int fromLineNum){
// valid log file
if (logFileName==null || logFileName.trim().length()==0) {
return new LogResult(fromLineNum, 0, "readLog fail, logFile not found", true);
}
File logFile = new File(logFileName);
if (!logFile.exists()) {
return new LogResult(fromLineNum, 0, "readLog fail, logFile not exists", true);
}
// read file
StringBuffer logContentBuffer = new StringBuffer();
int toLineNum = 0;
LineNumberReader reader = null;
try {
//reader = new LineNumberReader(new FileReader(logFile));
reader = new LineNumberReader(new InputStreamReader(new FileInputStream(logFile), "utf-8"));
String line = null;
while ((line = reader.readLine())!=null) {
toLineNum = reader.getLineNumber(); // [from, to], start as 1
if (toLineNum >= fromLineNum) {
logContentBuffer.append(line).append("\n");
}
}
} catch (IOException e) {
logger.error(e.getMessage(), e);
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
// result
LogResult logResult = new LogResult(fromLineNum, toLineNum, logContentBuffer.toString(), false);
return logResult;
/*
// it will return the number of characters actually skipped
reader.skip(Long.MAX_VALUE);
int maxLineNum = reader.getLineNumber();
maxLineNum++; // 最大行号
*/
}
/**
* read log data
* @param logFile
* @return log line content
*/
public static String readLines(File logFile){
BufferedReader reader = null;
try {
reader = new BufferedReader(new InputStreamReader(new FileInputStream(logFile), "utf-8"));
if (reader != null) {
StringBuilder sb = new StringBuilder();
String line = null;
while ((line = reader.readLine()) != null) {
sb.append(line).append("\n");
}
return sb.toString();
}
} catch (IOException e) {
logger.error(e.getMessage(), e);
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
return null;
}
}

View File

@@ -0,0 +1,257 @@
package com.xxl.job.core.server;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.impl.ExecutorBizImpl;
import com.xxl.job.core.biz.model.*;
import com.xxl.job.core.thread.ExecutorRegistryThread;
import com.xxl.job.core.util.GsonTool;
import com.xxl.job.core.util.ThrowableUtil;
import com.xxl.job.core.util.XxlJobRemotingUtil;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.http.*;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.concurrent.*;
/**
* Copy from : https://github.com/xuxueli/xxl-rpc
*
* @author xuxueli 2020-04-11 21:25
*/
public class EmbedServer {
private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);
private ExecutorBiz executorBiz;
private Thread thread;
public void start(final String address, final int port, final String appname, final String accessToken) {
executorBiz = new ExecutorBizImpl();
thread = new Thread(new Runnable() {
@Override
public void run() {
// param
EventLoopGroup bossGroup = new NioEventLoopGroup();
EventLoopGroup workerGroup = new NioEventLoopGroup();
ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
0,
200,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(2000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "xxl-job, EmbedServer bizThreadPool-" + r.hashCode());
}
},
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
}
});
try {
// start server
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<SocketChannel>() {
@Override
public void initChannel(SocketChannel channel) throws Exception {
channel.pipeline()
.addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS)) // beat 3N, close if idle
.addLast(new HttpServerCodec())
.addLast(new HttpObjectAggregator(5 * 1024 * 1024)) // merge request & reponse to FULL
.addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));
}
})
.childOption(ChannelOption.SO_KEEPALIVE, true);
// bind
ChannelFuture future = bootstrap.bind(port).sync();
logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);
// start registry
startRegistry(appname, address);
// wait util stop
future.channel().closeFuture().sync();
} catch (InterruptedException e) {
logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
} finally {
// stop
try {
workerGroup.shutdownGracefully();
bossGroup.shutdownGracefully();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
}
});
thread.setDaemon(true); // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
thread.start();
}
public void stop() throws Exception {
// destroy server thread
if (thread != null && thread.isAlive()) {
thread.interrupt();
}
// stop registry
stopRegistry();
logger.info(">>>>>>>>>>> xxl-job remoting server destroy success.");
}
// ---------------------- registry ----------------------
/**
* netty_http
* <p>
* Copy from : https://github.com/xuxueli/xxl-rpc
*
* @author xuxueli 2015-11-24 22:25:15
*/
public static class EmbedHttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {
private static final Logger logger = LoggerFactory.getLogger(EmbedHttpServerHandler.class);
private ExecutorBiz executorBiz;
private String accessToken;
private ThreadPoolExecutor bizThreadPool;
public EmbedHttpServerHandler(ExecutorBiz executorBiz, String accessToken, ThreadPoolExecutor bizThreadPool) {
this.executorBiz = executorBiz;
this.accessToken = accessToken;
this.bizThreadPool = bizThreadPool;
}
@Override
protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
// request parse
//final byte[] requestBytes = ByteBufUtil.getBytes(msg.content()); // byteBuf.toString(io.netty.util.CharsetUtil.UTF_8);
String requestData = msg.content().toString(CharsetUtil.UTF_8);
String uri = msg.uri();
HttpMethod httpMethod = msg.method();
boolean keepAlive = HttpUtil.isKeepAlive(msg);
String accessTokenReq = msg.headers().get(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN);
// invoke
bizThreadPool.execute(new Runnable() {
@Override
public void run() {
// do invoke
Object responseObj = process(httpMethod, uri, requestData, accessTokenReq);
// to json
String responseJson = GsonTool.toJson(responseObj);
// write response
writeResponse(ctx, keepAlive, responseJson);
}
});
}
private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
// valid
if (HttpMethod.POST != httpMethod) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, HttpMethod not support.");
}
if (uri == null || uri.trim().length() == 0) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty.");
}
if (accessToken != null
&& accessToken.trim().length() > 0
&& !accessToken.equals(accessTokenReq)) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
}
// services mapping
try {
switch (uri) {
case "/beat":
return executorBiz.beat();
case "/idleBeat":
IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
return executorBiz.idleBeat(idleBeatParam);
case "/run":
TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
return executorBiz.run(triggerParam);
case "/kill":
KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
return executorBiz.kill(killParam);
case "/log":
LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
return executorBiz.log(logParam);
default:
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping(" + uri + ") not found.");
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
return new ReturnT<String>(ReturnT.FAIL_CODE, "request error:" + ThrowableUtil.toString(e));
}
}
/**
* write response
*/
private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, String responseJson) {
// write response
FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK, Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8)); // Unpooled.wrappedBuffer(responseJson)
response.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/html;charset=UTF-8"); // HttpHeaderValues.TEXT_PLAIN.toString()
response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
if (keepAlive) {
response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
}
ctx.writeAndFlush(response);
}
@Override
public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
ctx.flush();
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
logger.error(">>>>>>>>>>> xxl-job provider netty_http server caught exception", cause);
ctx.close();
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
if (evt instanceof IdleStateEvent) {
ctx.channel().close(); // beat 3N, close if idle
logger.debug(">>>>>>>>>>> xxl-job provider netty_http server close an idle channel.");
} else {
super.userEventTriggered(ctx, evt);
}
}
}
// ---------------------- registry ----------------------
public void startRegistry(final String appname, final String address) {
// start registry
ExecutorRegistryThread.getInstance().start(appname, address);
}
public void stopRegistry() {
// stop registry
ExecutorRegistryThread.getInstance().toStop();
}
}

View File

@@ -0,0 +1,130 @@
package com.xxl.job.core.thread;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.RegistryParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.enums.RegistryConfig;
import com.xxl.job.core.executor.XxlJobExecutor;
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;
}
registryThread = new Thread(new Runnable() {
@Override
public void run() {
// registry
while (!toStop) {
try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
try {
ReturnT<String> registryResult = adminBiz.registry(registryParam);
if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
registryResult = ReturnT.SUCCESS;
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 (Exception e) {
logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
try {
if (!toStop) {
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
}
} catch (InterruptedException e) {
if (!toStop) {
logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());
}
}
}
// registry remove
try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
try {
ReturnT<String> registryResult = adminBiz.registryRemove(registryParam);
if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
registryResult = ReturnT.SUCCESS;
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 (Exception e) {
if (!toStop) {
logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e);
}
}
}
} catch (Exception 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 (InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
}
}

View File

@@ -0,0 +1,125 @@
package com.xxl.job.core.thread;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.util.FileUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.text.ParseException;
import java.text.SimpleDateFormat;
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;
}
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();
for (File childFile: childDirs) {
// valid
if (!childFile.isDirectory()) {
continue;
}
if (childFile.getName().indexOf("-") == -1) {
continue;
}
// file create date
Date logFileCreateDate = null;
try {
SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
logFileCreateDate = simpleDateFormat.parse(childFile.getName());
} catch (ParseException e) {
logger.error(e.getMessage(), e);
}
if (logFileCreateDate == null) {
continue;
}
if ((todayDate.getTime()-logFileCreateDate.getTime()) >= logRetentionDays * (24 * 60 * 60 * 1000) ) {
FileUtil.deleteRecursively(childFile);
}
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
try {
TimeUnit.DAYS.sleep(1);
} catch (InterruptedException 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);
}
}
}

View File

@@ -0,0 +1,256 @@
package com.xxl.job.core.thread;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
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.handler.IJobHandler;
import com.xxl.job.core.log.XxlJobFileAppender;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.PrintWriter;
import java.io.StringWriter;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.*;
/**
* handler thread
*
* @author xuxueli 2016-1-16 19:52:47
*/
public class JobThread extends Thread {
private static Logger logger = LoggerFactory.getLogger(JobThread.class);
private String jobId;
private IJobHandler handler;
private LinkedBlockingQueue<TriggerParam> triggerQueue;
private Set<String> triggerLogIdSet; // avoid repeat trigger for the same TRIGGER_LOG_ID
private volatile boolean toStop = false;
private String stopReason;
private boolean running = false; // if running job
private int idleTimes = 0; // idel times
public JobThread(String jobId, IJobHandler handler) {
this.jobId = jobId;
this.handler = handler;
this.triggerQueue = new LinkedBlockingQueue<TriggerParam>();
this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<String>());
// assign job thread name
this.setName("xxl-job, JobThread-" + jobId + "-" + System.currentTimeMillis());
}
public IJobHandler getHandler() {
return handler;
}
/**
* new trigger to queue
*
* @param triggerParam
* @return
*/
public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
// avoid repeat
if (triggerLogIdSet.contains(triggerParam.getLogId())) {
logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
}
triggerLogIdSet.add(triggerParam.getLogId());
triggerQueue.add(triggerParam);
return ReturnT.SUCCESS;
}
/**
* kill job thread
*
* @param stopReason
*/
public void toStop(String stopReason) {
/**
* Thread.interrupt只支持终止线程的阻塞状态(wait、join、sleep)
* 在阻塞出抛出InterruptedException异常,但是并不会终止运行的线程本身;
* 所以需要注意,此处彻底销毁本线程,需要通过共享变量方式;
*/
this.toStop = true;
this.stopReason = stopReason;
}
/**
* is running job
*
* @return
*/
public boolean isRunningOrHasQueue() {
return running || triggerQueue.size() > 0;
}
@Override
public void run() {
// init
try {
handler.init();
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
// execute
while (!toStop) {
running = false;
idleTimes++;
TriggerParam triggerParam = null;
try {
// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand
// of poll(timeout)
triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
if (triggerParam != null) {
running = true;
idleTimes = 0;
triggerLogIdSet.remove(triggerParam.getLogId());
// log filename, like "logPath/yyyy-MM-dd/9999.log"
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTime()),
triggerParam.getLogId());
XxlJobContext xxlJobContext = new XxlJobContext(
triggerParam.getJobId(),
triggerParam.getExecutorParams(),
logFileName,
triggerParam.getBroadcastIndex(),
triggerParam.getBroadcastTotal());
// init job context
XxlJobContext.setXxlJobContext(xxlJobContext);
// execute
XxlJobHelper.log("<br>----------- xxl-job job execute start -----------<br>----------- Param:"
+ xxlJobContext.getJobParam());
if (triggerParam.getExecutorTimeout() > 0) {
// limit timeout
Thread futureThread = null;
try {
FutureTask<Boolean> futureTask = new FutureTask<Boolean>(new Callable<Boolean>() {
@Override
public Boolean call() throws Exception {
// init job context
XxlJobContext.setXxlJobContext(xxlJobContext);
handler.execute();
return true;
}
});
futureThread = new Thread(futureTask);
futureThread.start();
futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);
} catch (TimeoutException e) {
XxlJobHelper.log("<br>----------- xxl-job job execute timeout");
XxlJobHelper.log(e);
// handle result
XxlJobHelper.handleTimeout("job execute timeout ");
} finally {
if (futureThread != null) {
futureThread.interrupt();
}
}
} else {
// just execute
handler.execute();
}
// valid execute handle data
if (XxlJobContext.getXxlJobContext().getHandleCode() <= 0) {
XxlJobHelper.handleFail("job handle result lost.");
} else {
String tempHandleMsg = XxlJobContext.getXxlJobContext().getHandleMsg();
tempHandleMsg = (tempHandleMsg != null && tempHandleMsg.length() > 50000)
? tempHandleMsg.substring(0, 50000).concat("...")
: tempHandleMsg;
XxlJobContext.getXxlJobContext().setHandleMsg(tempHandleMsg);
}
XxlJobHelper.log(
"<br>----------- xxl-job job execute end(finish) -----------<br>----------- Result: handleCode="
+ XxlJobContext.getXxlJobContext().getHandleCode()
+ ", handleMsg = "
+ XxlJobContext.getXxlJobContext().getHandleMsg());
} else {
if (idleTimes > 30) {
if (triggerQueue.size() == 0) { // avoid concurrent trigger causes jobId-lost
XxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");
}
}
}
} catch (Throwable e) {
if (toStop) {
XxlJobHelper.log("<br>----------- JobThread toStop, stopReason:" + stopReason);
}
// handle result
StringWriter stringWriter = new StringWriter();
e.printStackTrace(new PrintWriter(stringWriter));
String errorMsg = stringWriter.toString();
XxlJobHelper.handleFail(errorMsg);
XxlJobHelper.log("<br>----------- JobThread Exception:" + errorMsg
+ "<br>----------- xxl-job job execute end(error) -----------");
} finally {
if (triggerParam != null) {
// callback handler info
if (!toStop) {
// commonm
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.getXxlJobContext().getHandleCode(),
XxlJobContext.getXxlJobContext().getHandleMsg()));
} else {
// is killed
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.HANDLE_CODE_FAIL,
stopReason + " [job running, killed]"));
}
}
}
}
// callback trigger request in queue
while (triggerQueue != null && triggerQueue.size() > 0) {
TriggerParam triggerParam = triggerQueue.poll();
if (triggerParam != null) {
// is killed
TriggerCallbackThread.pushCallBack(new HandleCallbackParam(
triggerParam.getLogId(),
triggerParam.getLogDateTime(),
XxlJobContext.HANDLE_CODE_FAIL,
stopReason + " [job not executed, in the job queue, killed.]"));
}
}
// destroy
try {
handler.destroy();
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
logger.info(">>>>>>>>>>> xxl-job JobThread stoped, hashCode:{}", Thread.currentThread());
}
}

View File

@@ -0,0 +1,271 @@
package com.xxl.job.core.thread;
import com.xxl.job.core.biz.AdminBiz;
import com.xxl.job.core.biz.model.HandleCallbackParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.context.XxlJobContext;
import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.enums.RegistryConfig;
import com.xxl.job.core.executor.XxlJobExecutor;
import com.xxl.job.core.log.XxlJobFileAppender;
import com.xxl.job.core.util.FileUtil;
import com.xxl.job.core.util.JdkSerializeTool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
/**
* Created by xuxueli on 16/7/22.
*/
public class TriggerCallbackThread {
private static Logger logger = LoggerFactory.getLogger(TriggerCallbackThread.class);
private static TriggerCallbackThread instance = new TriggerCallbackThread();
public static TriggerCallbackThread getInstance() {
return instance;
}
/**
* job results callback queue
*/
private LinkedBlockingQueue<HandleCallbackParam> callBackQueue = new LinkedBlockingQueue<HandleCallbackParam>();
public static void pushCallBack(HandleCallbackParam 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;
}
// callback
triggerCallbackThread = new Thread(new Runnable() {
@Override
public void run() {
// normal callback
while (!toStop) {
try {
HandleCallbackParam callback = getInstance().callBackQueue.take();
if (callback != null) {
// callback list param
List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
getInstance().callBackQueue.drainTo(callbackParamList);
callbackParamList.add(callback);
// callback, will retry if error
if (callbackParamList != null && callbackParamList.size() > 0) {
doCallback(callbackParamList);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
}
// last callback
try {
List<HandleCallbackParam> callbackParamList = new ArrayList<HandleCallbackParam>();
getInstance().callBackQueue.drainTo(callbackParamList);
if (callbackParamList != null && callbackParamList.size() > 0) {
doCallback(callbackParamList);
}
} catch (Exception 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();
// retry
triggerRetryCallbackThread = new Thread(new Runnable() {
@Override
public void run() {
while (!toStop) {
try {
retryFailCallbackFile();
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
try {
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
} catch (InterruptedException e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, executor retry callback thread destroy.");
}
});
triggerRetryCallbackThread.setDaemon(true);
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 (InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
// stop retry, interrupt and wait
if (triggerRetryCallbackThread != null) {
triggerRetryCallbackThread.interrupt();
try {
triggerRetryCallbackThread.join();
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
}
/**
* do callback, will retry if error
*
* @param callbackParamList
*/
private void doCallback(List<HandleCallbackParam> callbackParamList) {
boolean callbackRet = false;
// callback, will retry if error
for (AdminBiz adminBiz : XxlJobExecutor.getAdminBizList()) {
try {
ReturnT<String> callbackResult = adminBiz.callback(callbackParamList);
if (callbackResult != null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
callbackLog(callbackParamList, "<br>----------- xxl-job job callback finish.");
callbackRet = true;
break;
} else {
callbackLog(callbackParamList,
"<br>----------- xxl-job job callback fail, callbackResult:" + callbackResult);
}
} catch (Exception e) {
callbackLog(callbackParamList,
"<br>----------- xxl-job job callback error, errorMsg:" + e.getMessage());
}
}
if (!callbackRet) {
appendFailCallbackFile(callbackParamList);
}
}
/**
* callback log
*/
private void callbackLog(List<HandleCallbackParam> callbackParamList, String logContent) {
for (HandleCallbackParam callbackParam : callbackParamList) {
String logFileName = XxlJobFileAppender.makeLogFileName(new Date(callbackParam.getLogDateTim()),
callbackParam.getLogId());
XxlJobContext.setXxlJobContext(new XxlJobContext(
"0",
null,
logFileName,
-1,
-1));
XxlJobHelper.log(logContent);
}
}
// ---------------------- fail-callback file ----------------------
private static String failCallbackFilePath = XxlJobFileAppender.getLogPath().concat(File.separator)
.concat("callbacklog").concat(File.separator);
private static String failCallbackFileName = failCallbackFilePath.concat("xxl-job-callback-{x}").concat(".log");
private void appendFailCallbackFile(List<HandleCallbackParam> callbackParamList) {
// valid
if (callbackParamList == null || callbackParamList.size() == 0) {
return;
}
// append file
byte[] callbackParamList_bytes = JdkSerializeTool.serialize(callbackParamList);
File callbackLogFile = new File(
failCallbackFileName.replace("{x}", String.valueOf(System.currentTimeMillis())));
if (callbackLogFile.exists()) {
for (int i = 0; i < 100; i++) {
callbackLogFile = new File(failCallbackFileName.replace("{x}",
String.valueOf(System.currentTimeMillis()).concat("-").concat(String.valueOf(i))));
if (!callbackLogFile.exists()) {
break;
}
}
}
FileUtil.writeFileContent(callbackLogFile, callbackParamList_bytes);
}
private void retryFailCallbackFile() {
// valid
File callbackLogPath = new File(failCallbackFilePath);
if (!callbackLogPath.exists()) {
return;
}
if (callbackLogPath.isFile()) {
callbackLogPath.delete();
}
if (!(callbackLogPath.isDirectory() && callbackLogPath.list() != null && callbackLogPath.list().length > 0)) {
return;
}
// load and clear file, retry
for (File callbaclLogFile : callbackLogPath.listFiles()) {
byte[] callbackParamList_bytes = FileUtil.readFileContent(callbaclLogFile);
// avoid empty file
if (callbackParamList_bytes == null || callbackParamList_bytes.length < 1) {
callbaclLogFile.delete();
continue;
}
@SuppressWarnings("unchecked")
List<HandleCallbackParam> callbackParamList = (List<HandleCallbackParam>) JdkSerializeTool
.deserialize(callbackParamList_bytes, List.class);
callbaclLogFile.delete();
doCallback(callbackParamList);
}
}
}

View File

@@ -0,0 +1,157 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
/**
* date util
*
* @author xuxueli 2018-08-19 01:24:11
*/
public class DateUtil {
// ---------------------- format parse ----------------------
private static Logger logger = LoggerFactory.getLogger(DateUtil.class);
private static final String DATE_FORMAT = "yyyy-MM-dd";
private static final String DATETIME_FORMAT = "yyyy-MM-dd HH:mm:ss";
private static final ThreadLocal<Map<String, DateFormat>> dateFormatThreadLocal = new ThreadLocal<Map<String, DateFormat>>();
private static DateFormat getDateFormat(String pattern) {
if (pattern==null || pattern.trim().length()==0) {
throw new IllegalArgumentException("pattern cannot be empty.");
}
Map<String, DateFormat> dateFormatMap = dateFormatThreadLocal.get();
if(dateFormatMap!=null && dateFormatMap.containsKey(pattern)){
return dateFormatMap.get(pattern);
}
synchronized (dateFormatThreadLocal) {
if (dateFormatMap == null) {
dateFormatMap = new HashMap<String, DateFormat>();
}
dateFormatMap.put(pattern, new SimpleDateFormat(pattern));
dateFormatThreadLocal.set(dateFormatMap);
}
return dateFormatMap.get(pattern);
}
/**
* format datetime. like "yyyy-MM-dd"
*
* @param date
* @return
* @throws ParseException
*/
public static String formatDate(Date date) {
return format(date, DATE_FORMAT);
}
/**
* format date. like "yyyy-MM-dd HH:mm:ss"
*
* @param date
* @return
* @throws ParseException
*/
public static String formatDateTime(Date date) {
return format(date, DATETIME_FORMAT);
}
/**
* format date
*
* @param date
* @param patten
* @return
* @throws ParseException
*/
public static String format(Date date, String patten) {
return getDateFormat(patten).format(date);
}
/**
* parse date string, like "yyyy-MM-dd HH:mm:s"
*
* @param dateString
* @return
* @throws ParseException
*/
public static Date parseDate(String dateString){
return parse(dateString, DATE_FORMAT);
}
/**
* parse datetime string, like "yyyy-MM-dd HH:mm:ss"
*
* @param dateString
* @return
* @throws ParseException
*/
public static Date parseDateTime(String dateString) {
return parse(dateString, DATETIME_FORMAT);
}
/**
* parse date
*
* @param dateString
* @param pattern
* @return
* @throws ParseException
*/
public static Date parse(String dateString, String pattern) {
try {
Date date = getDateFormat(pattern).parse(dateString);
return date;
} catch (Exception e) {
logger.warn("parse date error, dateString = {}, pattern={}; errorMsg = {}", dateString, pattern, e.getMessage());
return null;
}
}
// ---------------------- add date ----------------------
public static Date addYears(final Date date, final int amount) {
return add(date, Calendar.YEAR, amount);
}
public static Date addMonths(final Date date, final int amount) {
return add(date, Calendar.MONTH, amount);
}
public static Date addDays(final Date date, final int amount) {
return add(date, Calendar.DAY_OF_MONTH, amount);
}
public static Date addHours(final Date date, final int amount) {
return add(date, Calendar.HOUR_OF_DAY, amount);
}
public static Date addMinutes(final Date date, final int amount) {
return add(date, Calendar.MINUTE, amount);
}
private static Date add(final Date date, final int calendarField, final int amount) {
if (date == null) {
return null;
}
final Calendar c = Calendar.getInstance();
c.setTime(date);
c.add(calendarField, amount);
return c.getTime();
}
}

View File

@@ -0,0 +1,182 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileOutputStream;
import java.io.IOException;
/**
* file tool
*
* @author xuxueli 2017-12-29 17:56:48
*/
public class FileUtil {
private static Logger logger = LoggerFactory.getLogger(FileUtil.class);
/**
* delete recursively
*
* @param root
* @return
*/
public static boolean deleteRecursively(File root) {
if (root != null && root.exists()) {
if (root.isDirectory()) {
File[] children = root.listFiles();
if (children != null) {
for (File child : children) {
deleteRecursively(child);
}
}
}
return root.delete();
}
return false;
}
public static void deleteFile(String fileName) {
// file
File file = new File(fileName);
if (file.exists()) {
file.delete();
}
}
public static void writeFileContent(File file, byte[] data) {
// file
if (!file.exists()) {
file.getParentFile().mkdirs();
}
// append file content
FileOutputStream fos = null;
try {
fos = new FileOutputStream(file);
fos.write(data);
fos.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (fos != null) {
try {
fos.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
public static byte[] readFileContent(File file) {
Long filelength = file.length();
byte[] filecontent = new byte[filelength.intValue()];
FileInputStream in = null;
try {
in = new FileInputStream(file);
in.read(filecontent);
in.close();
return filecontent;
} catch (Exception e) {
logger.error(e.getMessage(), e);
return null;
} finally {
if (in != null) {
try {
in.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
/*public static void appendFileLine(String fileName, String content) {
// file
File file = new File(fileName);
if (!file.exists()) {
try {
file.createNewFile();
} catch (IOException e) {
logger.error(e.getMessage(), e);
return;
}
}
// content
if (content == null) {
content = "";
}
content += "\r\n";
// append file content
FileOutputStream fos = null;
try {
fos = new FileOutputStream(file, true);
fos.write(content.getBytes("utf-8"));
fos.flush();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (fos != null) {
try {
fos.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
}
public static List<String> loadFileLines(String fileName){
List<String> result = new ArrayList<>();
// valid log file
File file = new File(fileName);
if (!file.exists()) {
return result;
}
// read file
StringBuffer logContentBuffer = new StringBuffer();
int toLineNum = 0;
LineNumberReader reader = null;
try {
//reader = new LineNumberReader(new FileReader(logFile));
reader = new LineNumberReader(new InputStreamReader(new FileInputStream(file), "utf-8"));
String line = null;
while ((line = reader.readLine())!=null) {
if (line!=null && line.trim().length()>0) {
result.add(line);
}
}
} catch (IOException e) {
logger.error(e.getMessage(), e);
} finally {
if (reader != null) {
try {
reader.close();
} catch (IOException e) {
logger.error(e.getMessage(), e);
}
}
}
return result;
}*/
}

View File

@@ -0,0 +1,94 @@
package com.xxl.job.core.util;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.reflect.TypeToken;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.List;
/**
* @author xuxueli 2020-04-11 20:56:31
*/
public class GsonTool {
private static Gson gson = null;
static {
gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
}
/**
* Object 转成 json
*
* @param src
* @return String
*/
public static String toJson(Object src) {
return gson.toJson(src);
}
/**
* json 转成 特定的cls的Object
*
* @param json
* @param classOfT
* @return
*/
public static <T> T fromJson(String json, Class<T> classOfT) {
return gson.fromJson(json, classOfT);
}
/**
* json 转成 特定的 rawClass<classOfT> 的 Object
*
* @param json
* @param classOfT
* @param argClassOfT
* @return
*/
public static <T> T fromJson(String json, Class<T> classOfT, Class<?> argClassOfT) {
Type type = new ParameterizedType4ReturnT(classOfT, new Class[] { argClassOfT });
return gson.fromJson(json, type);
}
public static class ParameterizedType4ReturnT implements ParameterizedType {
private final Class<?> raw;
private final Type[] args;
public ParameterizedType4ReturnT(Class<?> raw, Type[] args) {
this.raw = raw;
this.args = args != null ? args : new Type[0];
}
@Override
public Type[] getActualTypeArguments() {
return args;
}
@Override
public Type getRawType() {
return raw;
}
@Override
public Type getOwnerType() {
return null;
}
}
/**
* json 转成 特定的cls的list
*
* @param json
* @param classOfT
* @return
*/
public static <T> List<T> fromJsonList(String json, Class<T> classOfT) {
return gson.fromJson(
json,
new TypeToken<List<T>>() {
}.getType());
}
}

View File

@@ -0,0 +1,204 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.Inet6Address;
import java.net.InetAddress;
import java.net.NetworkInterface;
import java.net.UnknownHostException;
import java.util.Enumeration;
import java.util.regex.Pattern;
/**
* ip tool
*
* @author xuxueli 2016-5-22 11:38:05
*/
public class IpUtil {
private static final Logger logger = LoggerFactory.getLogger(IpUtil.class);
private static final String ANYHOST_VALUE = "0.0.0.0";
private static final String LOCALHOST_VALUE = "127.0.0.1";
private static final Pattern IP_PATTERN = Pattern.compile("\\d{1,3}(\\.\\d{1,3}){3,5}$");
private static volatile InetAddress LOCAL_ADDRESS = null;
// ---------------------- valid ----------------------
private static InetAddress toValidAddress(InetAddress address) {
if (address instanceof Inet6Address) {
Inet6Address v6Address = (Inet6Address) address;
if (isPreferIPV6Address()) {
return normalizeV6Address(v6Address);
}
}
if (isValidV4Address(address)) {
return address;
}
return null;
}
private static boolean isPreferIPV6Address() {
return Boolean.getBoolean("java.net.preferIPv6Addresses");
}
/**
* valid Inet4Address
*
* @param address
* @return
*/
private static boolean isValidV4Address(InetAddress address) {
if (address == null || address.isLoopbackAddress()) {
return false;
}
String name = address.getHostAddress();
boolean result = (name != null
&& IP_PATTERN.matcher(name).matches()
&& !ANYHOST_VALUE.equals(name)
&& !LOCALHOST_VALUE.equals(name));
return result;
}
/**
* normalize the ipv6 Address, convert scope name to scope id.
* e.g.
* convert
* fe80:0:0:0:894:aeec:f37d:23e1%en0
* to
* fe80:0:0:0:894:aeec:f37d:23e1%5
* <p>
* The %5 after ipv6 address is called scope id.
* see java doc of {@link Inet6Address} for more details.
*
* @param address the input address
* @return the normalized address, with scope id converted to int
*/
private static InetAddress normalizeV6Address(Inet6Address address) {
String addr = address.getHostAddress();
int i = addr.lastIndexOf('%');
if (i > 0) {
try {
return InetAddress.getByName(addr.substring(0, i) + '%' + address.getScopeId());
} catch (UnknownHostException e) {
// ignore
logger.debug("Unknown IPV6 address: ", e);
}
}
return address;
}
// ---------------------- find ip ----------------------
private static InetAddress getLocalAddress0() {
InetAddress localAddress = null;
try {
localAddress = InetAddress.getLocalHost();
InetAddress addressItem = toValidAddress(localAddress);
if (addressItem != null) {
return addressItem;
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
try {
Enumeration<NetworkInterface> interfaces = NetworkInterface.getNetworkInterfaces();
if (null == interfaces) {
return localAddress;
}
while (interfaces.hasMoreElements()) {
try {
NetworkInterface network = interfaces.nextElement();
if (network.isLoopback() || network.isVirtual() || !network.isUp()) {
continue;
}
Enumeration<InetAddress> addresses = network.getInetAddresses();
while (addresses.hasMoreElements()) {
try {
InetAddress addressItem = toValidAddress(addresses.nextElement());
if (addressItem != null) {
try {
if(addressItem.isReachable(100)){
return addressItem;
}
} catch (IOException e) {
// ignore
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
}
} catch (Throwable e) {
logger.error(e.getMessage(), e);
}
return localAddress;
}
// ---------------------- tool ----------------------
/**
* Find first valid IP from local network card
*
* @return first valid local IP
*/
public static InetAddress getLocalAddress() {
if (LOCAL_ADDRESS != null) {
return LOCAL_ADDRESS;
}
InetAddress localAddress = getLocalAddress0();
LOCAL_ADDRESS = localAddress;
return localAddress;
}
/**
* get ip address
*
* @return String
*/
public static String getIp(){
return getLocalAddress().getHostAddress();
}
/**
* get ip:port
*
* @param port
* @return String
*/
public static String getIpPort(int port){
String ip = getIp();
return getIpPort(ip, port);
}
public static String getIpPort(String ip, int port){
if (ip==null) {
return null;
}
return ip.concat(":").concat(String.valueOf(port));
}
public static Object[] parseIpPort(String address){
String[] array = address.split(":");
String host = array[0];
int port = Integer.parseInt(array[1]);
return new Object[]{host, port};
}
}

View File

@@ -0,0 +1,51 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.*;
/**
* @author xuxueli 2020-04-12 0:14:00
*/
public class JdkSerializeTool {
private static Logger logger = LoggerFactory.getLogger(JdkSerializeTool.class);
// ------------------------ serialize and unserialize ------------------------
/**
* 将对象-->byte[] (由于jedis中不支持直接存储object所以转换成byte[]存入)
*
* @param object
* @return
*/
public static byte[] serialize(Object object) {
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
ObjectOutputStream oos = new ObjectOutputStream(baos)) {
// 序列化
oos.writeObject(object);
return baos.toByteArray();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return null;
}
/**
* 将byte[] -->Object
*
* @param bytes
* @return
*/
public static <T> Object deserialize(byte[] bytes, Class<T> clazz) {
try (ByteArrayInputStream bais = new ByteArrayInputStream(bytes);
ObjectInputStream ois = new ObjectInputStream(bais)) {
// 反序列化
return ois.readObject();
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
return null;
}
}

View File

@@ -0,0 +1,71 @@
package com.xxl.job.core.util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.net.ServerSocket;
/**
* net util
*
* @author xuxueli 2017-11-29 17:00:25
*/
public class NetUtil {
private static Logger logger = LoggerFactory.getLogger(NetUtil.class);
/**
* find avaliable port
*
* @param defaultPort
* @return
*/
public static int findAvailablePort(int defaultPort) {
int portTmp = defaultPort;
while (portTmp < 65535) {
if (!isPortUsed(portTmp)) {
return portTmp;
} else {
portTmp++;
}
}
portTmp = defaultPort--;
while (portTmp > 0) {
if (!isPortUsed(portTmp)) {
return portTmp;
} else {
portTmp--;
}
}
throw new RuntimeException("no available port.");
}
/**
* check port used
*
* @param port
* @return
*/
public static boolean isPortUsed(int port) {
boolean used = false;
ServerSocket serverSocket = null;
try {
serverSocket = new ServerSocket(port);
used = false;
} catch (IOException e) {
logger.info(">>>>>>>>>>> xxl-job, port[{}] is in use.", port);
used = true;
} finally {
if (serverSocket != null) {
try {
serverSocket.close();
} catch (IOException e) {
logger.info("");
}
}
}
return used;
}
}

View File

@@ -0,0 +1,235 @@
package com.xxl.job.core.util;
import com.xxl.job.core.context.XxlJobHelper;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
import java.util.List;
/**
* 1、内嵌编译器如"PythonInterpreter"无法引用扩展包因此推荐使用java调用控制台进程方式"Runtime.getRuntime().exec()"来运行脚本(shell或python)
* 2、因为通过java调用控制台进程方式实现需要保证目标机器PATH路径正确配置对应编译器
* 3、暂时脚本执行日志只能在脚本执行结束后一次性获取无法保证实时性因此为确保日志实时性可改为将脚本打印的日志存储在指定的日志文件上
* 4、python 异常输出优先级高于标准输出体现在Log文件中因此推荐通过logging方式打日志保持和异常信息一致否则用prinf日志顺序会错乱
*
* Created by xuxueli on 17/2/25.
*/
public class ScriptUtil {
/**
* make script file
*
* @param scriptFileName
* @param content
* @throws IOException
*/
public static void markScriptFile(String scriptFileName, String content) throws IOException {
// make file, filePath/gluesource/666-123456789.py
FileOutputStream fileOutputStream = null;
try {
fileOutputStream = new FileOutputStream(scriptFileName);
fileOutputStream.write(content.getBytes("UTF-8"));
fileOutputStream.close();
} catch (Exception e) {
throw e;
} finally {
if (fileOutputStream != null) {
fileOutputStream.close();
}
}
}
/**
* 脚本执行,日志文件实时输出
*
* @param command
* @param scriptFile
* @param logFile
* @param params
* @return
* @throws IOException
*/
public static int execToFile(String command, String scriptFile, String logFile, String... params)
throws IOException {
FileOutputStream fileOutputStream = null;
Thread inputThread = null;
Thread errThread = null;
try {
// file
fileOutputStream = new FileOutputStream(logFile, true);
// command
List<String> cmdarray = new ArrayList<>();
cmdarray.add(command);
cmdarray.add(scriptFile);
if (params != null && params.length > 0) {
for (String param : params) {
cmdarray.add(param);
}
}
String[] cmdarrayFinal = cmdarray.toArray(new String[cmdarray.size()]);
// process-exec
final Process process = Runtime.getRuntime().exec(cmdarrayFinal);
// log-thread
final FileOutputStream finalFileOutputStream = fileOutputStream;
inputThread = new Thread(new Runnable() {
@Override
public void run() {
try {
copy(process.getInputStream(), finalFileOutputStream, new byte[1024]);
} catch (IOException e) {
XxlJobHelper.log(e);
}
}
});
errThread = new Thread(new Runnable() {
@Override
public void run() {
try {
copy(process.getErrorStream(), finalFileOutputStream, new byte[1024]);
} catch (IOException e) {
XxlJobHelper.log(e);
}
}
});
inputThread.start();
errThread.start();
// process-wait
int exitValue = process.waitFor(); // exit code: 0=success, 1=error
// log-thread join
inputThread.join();
errThread.join();
return exitValue;
} catch (Exception e) {
XxlJobHelper.log(e);
return -1;
} finally {
if (fileOutputStream != null) {
try {
fileOutputStream.close();
} catch (IOException e) {
XxlJobHelper.log(e);
}
}
if (inputThread != null && inputThread.isAlive()) {
inputThread.interrupt();
}
if (errThread != null && errThread.isAlive()) {
errThread.interrupt();
}
}
}
/**
* 数据流CopyInput自动关闭Output不处理
*
* @param inputStream
* @param outputStream
* @param buffer
* @return
* @throws IOException
*/
private static long copy(InputStream inputStream, OutputStream outputStream, byte[] buffer) throws IOException {
try {
long total = 0;
for (;;) {
int res = inputStream.read(buffer);
if (res == -1) {
break;
}
if (res > 0) {
total += res;
if (outputStream != null) {
outputStream.write(buffer, 0, res);
}
}
}
if (outputStream != null) {
outputStream.flush();
}
// out = null;
inputStream.close();
inputStream = null;
return total;
} finally {
if (inputStream != null) {
inputStream.close();
}
}
}
/**
* 脚本执行,日志文件实时输出
*
* 优点:支持将目标数据实时输出到指定日志文件中去
* 缺点:
* 标准输出和错误输出优先级固定,可能和脚本中顺序不一致
* Java无法实时获取
*
* <!-- commons-exec -->
* <dependency>
* <groupId>org.apache.commons</groupId>
* <artifactId>commons-exec</artifactId>
* <version>${commons-exec.version}</version>
* </dependency>
*
* @param command
* @param scriptFile
* @param logFile
* @param params
* @return
* @throws IOException
*/
/*
* public static int execToFileB(String command, String scriptFile, String
* logFile, String... params) throws IOException {
* // 标准输出print null if watchdog timeout
* // 错误输出logging + 异常 still exists if watchdog timeout
* // 标准输入
*
* FileOutputStream fileOutputStream = null; //
* try {
* fileOutputStream = new FileOutputStream(logFile, true);
* PumpStreamHandler streamHandler = new PumpStreamHandler(fileOutputStream,
* fileOutputStream, null);
*
* // command
* CommandLine commandline = new CommandLine(command);
* commandline.addArgument(scriptFile);
* if (params!=null && params.length>0) {
* commandline.addArguments(params);
* }
*
* // exec
* DefaultExecutor exec = new DefaultExecutor();
* exec.setExitValues(null);
* exec.setStreamHandler(streamHandler);
* int exitValue = exec.execute(commandline); // exit code: 0=success, 1=error
* return exitValue;
* } catch (Exception e) {
* XxlJobLogger.log(e);
* return -1;
* } finally {
* if (fileOutputStream != null) {
* try {
* fileOutputStream.close();
* } catch (IOException e) {
* XxlJobLogger.log(e);
* }
*
* }
* }
* }
*/
}

View File

@@ -0,0 +1,72 @@
/**
* yunzhupaas Commercial License Agreement
* Copyright (c) 2018-2099, http://www.szlecheng.cn. All rights reserved.
* <p>
* Use of this software is governed by the Commercial License Agreement
* obtained after purchasing a license from yunzhupaas.
* <p>
* 1. This software is for development use only under a valid license
* from yunzhupaas.
* <p>
* 2. Redistribution of this software's source code to any third party
* without a commercial license is strictly prohibited.
* <p>
* 3. Licensees may copyright their own code but cannot use segments
* from this software for such purposes. Copyright of this software
* remains with yunzhupaas.
* <p>
* Using this software signifies agreement to this License, and the software
* must not be used for illegal purposes.
* <p>
* THIS yunzhupaasSoftWare IS PROVIDED "AS IS", WITHOUT WARRANTY. The author is
* not liable for any claims arising from secondary or illegal development.
* <p>
* Author: yunzhupaas.com
*/
// package com.xxl.job.core.util;
//
/// **
// * sharding vo
// * @author xuxueli 2017-07-25 21:26:38
// */
// public class ShardingUtil {
//
// private static InheritableThreadLocal<ShardingVO> contextHolder = new
// InheritableThreadLocal<ShardingVO>();
//
// public static class ShardingVO {
//
// private int index; // sharding index
// private int total; // sharding total
//
// public ShardingVO(int index, int total) {
// this.index = index;
// this.total = total;
// }
//
// public int getIndex() {
// return index;
// }
//
// public void setIndex(int index) {
// this.index = index;
// }
//
// public int getTotal() {
// return total;
// }
//
// public void setTotal(int total) {
// this.total = total;
// }
// }
//
// public static void setShardingVo(ShardingVO shardingVo){
// contextHolder.set(shardingVo);
// }
//
// public static ShardingVO getShardingVo(){
// return contextHolder.get();
// }
//
// }

View File

@@ -0,0 +1,25 @@
package com.xxl.job.core.util;
import java.io.PrintWriter;
import java.io.StringWriter;
/**
* @author xuxueli 2018-10-20 20:07:26
*/
public class ThrowableUtil {
/**
* parse error to string
*
* @param e
* @return
*/
public static String toString(Throwable e) {
StringWriter stringWriter = new StringWriter();
e.printStackTrace(new PrintWriter(stringWriter));
String errorMsg = stringWriter.toString();
return errorMsg;
}
}

View File

@@ -0,0 +1,167 @@
package com.xxl.job.core.util;
import com.xxl.job.core.biz.model.ReturnT;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import javax.net.ssl.*;
import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
import java.security.cert.CertificateException;
import java.security.cert.X509Certificate;
/**
* @author xuxueli 2018-11-25 00:55:31
*/
public class XxlJobRemotingUtil {
private static Logger logger = LoggerFactory.getLogger(XxlJobRemotingUtil.class);
public static final String XXL_JOB_ACCESS_TOKEN = "XXL-JOB-ACCESS-TOKEN";
// trust-https start
private static void trustAllHosts(HttpsURLConnection connection) {
try {
SSLContext sc = SSLContext.getInstance("TLS");
sc.init(null, trustAllCerts, new java.security.SecureRandom());
SSLSocketFactory newFactory = sc.getSocketFactory();
connection.setSSLSocketFactory(newFactory);
} catch (Exception e) {
logger.error("调度服务器连接失败:" + e.getMessage());
}
connection.setHostnameVerifier(new HostnameVerifier() {
@Override
public boolean verify(String hostname, SSLSession session) {
return true;
}
});
}
private static final TrustManager[] trustAllCerts = new TrustManager[] { new X509TrustManager() {
@Override
public java.security.cert.X509Certificate[] getAcceptedIssuers() {
return new java.security.cert.X509Certificate[] {};
}
@Override
public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
}
@Override
public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
}
} };
// trust-https end
/**
* post
*
* @param url
* @param accessToken
* @param timeout
* @param requestObj
* @param returnTargClassOfT
* @return
*/
public static <T> ReturnT<T> postBody(String url, String accessToken, int timeout, Object requestObj,
Class<T> returnTargClassOfT) {
HttpURLConnection connection = null;
BufferedReader bufferedReader = null;
try {
// connection
URL realUrl = new URL(url);
connection = (HttpURLConnection) realUrl.openConnection();
// trust-https
boolean useHttps = url.startsWith("https");
if (useHttps) {
HttpsURLConnection https = (HttpsURLConnection) connection;
trustAllHosts(https);
}
// connection setting
connection.setRequestMethod("POST");
connection.setDoOutput(true);
connection.setDoInput(true);
connection.setUseCaches(false);
connection.setReadTimeout(timeout * 1000);
connection.setConnectTimeout(3 * 1000);
connection.setRequestProperty("connection", "Keep-Alive");
connection.setRequestProperty("Content-Type", "application/json;charset=UTF-8");
connection.setRequestProperty("Accept-Charset", "application/json;charset=UTF-8");
if (accessToken != null && accessToken.trim().length() > 0) {
connection.setRequestProperty(XXL_JOB_ACCESS_TOKEN, accessToken);
}
// do connection
connection.connect();
// write requestBody
if (requestObj != null) {
String requestBody = GsonTool.toJson(requestObj);
DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
dataOutputStream.write(requestBody.getBytes("UTF-8"));
dataOutputStream.flush();
dataOutputStream.close();
}
/*
* byte[] requestBodyBytes = requestBody.getBytes("UTF-8");
* connection.setRequestProperty("Content-Length",
* String.valueOf(requestBodyBytes.length));
* OutputStream outwritestream = connection.getOutputStream();
* outwritestream.write(requestBodyBytes);
* outwritestream.flush();
* outwritestream.close();
*/
// valid StatusCode
int statusCode = connection.getResponseCode();
if (statusCode != 200) {
return new ReturnT<T>(ReturnT.FAIL_CODE,
"xxl-job remoting fail, StatusCode(" + statusCode + ") invalid. for url : " + url);
}
// result
bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), "UTF-8"));
StringBuilder result = new StringBuilder();
String line;
while ((line = bufferedReader.readLine()) != null) {
result.append(line);
}
String resultJson = result.toString();
// parse returnT
try {
@SuppressWarnings("unchecked")
ReturnT<T> returnT = (ReturnT<T>) GsonTool.fromJson(resultJson, ReturnT.class, returnTargClassOfT);
return returnT;
} catch (Exception e) {
logger.error("xxl-job remoting (url=" + url + ") response content invalid(" + resultJson + ").", e);
return new ReturnT<T>(ReturnT.FAIL_CODE,
"xxl-job remoting (url=" + url + ") response content invalid(" + resultJson + ").");
}
} catch (Exception e) {
logger.error("调度服务器连接失败:" + e.getMessage());
return new ReturnT<T>(ReturnT.FAIL_CODE,
"xxl-job remoting error(" + e.getMessage() + "), for url : " + url);
} finally {
try {
if (bufferedReader != null) {
bufferedReader.close();
}
if (connection != null) {
connection.disconnect();
}
} catch (Exception e2) {
logger.error(e2.getMessage());
}
}
}
}