From 4f9c0caaab671e26a00302a16740777744be38ad Mon Sep 17 00:00:00 2001
From: zhongrj <646384940@qq.com>
Date: Tue, 09 Jan 2024 21:50:19 +0800
Subject: [PATCH] 集成xxljob执行器
---
src/main/java/com/xxl/job/core/handler/impl/ScriptJobHandler.java | 93 +
src/main/java/com/xxl/job/core/handler/impl/MethodJobHandler.java | 53
src/main/java/com/xxl/job/core/util/ThrowableUtil.java | 24
src/main/java/com/xxl/job/core/util/JdkSerializeTool.java | 70
src/main/java/com/xxl/job/core/biz/model/IdleBeatParam.java | 29
src/main/java/org/springblade/xxljob/jobhandler/SampleXxlJob.java | 258 +++
src/main/resources/application-prod.yml | 14
src/main/java/com/xxl/job/core/context/XxlJobHelper.java | 252 +++
src/main/java/com/xxl/job/core/biz/client/AdminBizClient.java | 51
src/main/java/com/xxl/job/core/thread/JobLogFileCleanThread.java | 126 +
src/main/java/com/xxl/job/core/context/XxlJobContext.java | 119 +
src/main/java/com/xxl/job/core/util/ScriptUtil.java | 167 ++
src/main/java/com/xxl/job/core/enums/ExecutorBlockStrategyEnum.java | 37
src/main/java/com/xxl/job/core/util/DateUtil.java | 157 +
src/main/resources/application-dev.yml | 13
src/main/java/com/xxl/job/core/biz/model/ReturnT.java | 65
src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java | 279 +++
src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java | 212 ++
src/main/java/org/springblade/xxljob/config/XxlJobConfig.java | 59
src/main/java/com/xxl/job/core/util/FileUtil.java | 100 +
src/main/java/com/xxl/job/core/biz/model/TriggerParam.java | 144 +
src/main/java/com/xxl/job/core/handler/impl/GlueJobHandler.java | 40
src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java | 265 +++
src/main/java/com/xxl/job/core/util/XxlJobRemotingUtil.java | 158 +
src/main/java/com/xxl/job/core/handler/annotation/XxlJob.java | 30
src/main/resources/application-test.yml | 14
src/main/java/com/xxl/job/core/biz/ExecutorBiz.java | 49
src/main/java/com/xxl/job/core/enums/RegistryConfig.java | 13
src/main/java/com/xxl/job/core/handler/annotation/JobHandler.java | 24
src/main/java/com/xxl/job/core/biz/model/RegistryParam.java | 56
src/main/java/com/xxl/job/core/util/IpUtil.java | 184 ++
src/main/java/com/xxl/job/core/biz/model/LogParam.java | 48
src/main/java/com/xxl/job/core/glue/impl/SpringGlueFactory.java | 80
src/main/java/com/xxl/job/core/biz/model/KillParam.java | 29
src/main/java/com/xxl/job/core/biz/model/LogResult.java | 57
src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java | 172 ++
src/main/java/com/xxl/job/core/biz/AdminBiz.java | 42
src/main/java/com/xxl/job/core/util/NetUtil.java | 70
pom.xml | 15
src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java | 131 +
src/main/java/com/xxl/job/core/executor/impl/XxlJobSimpleExecutor.java | 75
src/main/java/com/xxl/job/core/executor/impl/XxlJobSpringExecutor.java | 123 +
src/main/java/com/xxl/job/core/handler/IJobHandler.java | 32
src/main/java/com/xxl/job/core/server/EmbedServer.java | 263 +++
src/main/java/com/xxl/job/core/util/GsonTool.java | 96 +
src/main/java/com/xxl/job/core/biz/client/ExecutorBizClient.java | 57
src/main/java/com/xxl/job/core/biz/model/HandleCallbackParam.java | 69
src/main/java/com/xxl/job/core/glue/GlueFactory.java | 90 +
src/main/java/com/xxl/job/core/thread/JobThread.java | 255 +++
src/main/java/com/xxl/job/core/glue/GlueTypeEnum.java | 53
50 files changed, 4,912 insertions(+), 0 deletions(-)
diff --git a/pom.xml b/pom.xml
index dc1b87c..911178d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -28,6 +28,9 @@
<docker.password>Harbor12345</docker.password>
<docker.namespace>blade</docker.namespace>
<docker.plugin.version>1.4.13</docker.plugin.version>
+
+ <netty-all.version>4.1.63.Final</netty-all.version>
+ <groovy.version>3.0.8</groovy.version>
</properties>
<dependencyManagement>
@@ -197,6 +200,18 @@
<artifactId>weixin-java-pay</artifactId>
<version>4.4.9.B</version>
</dependency>
+ <!--xxl-job-core 相关依赖-->
+ <dependency>
+ <groupId>io.netty</groupId>
+ <artifactId>netty-all</artifactId>
+ <version>${netty-all.version}</version>
+ </dependency>
+ <!-- groovy-all -->
+ <dependency>
+ <groupId>org.codehaus.groovy</groupId>
+ <artifactId>groovy</artifactId>
+ <version>${groovy.version}</version>
+ </dependency>
</dependencies>
<build>
diff --git a/src/main/java/com/xxl/job/core/biz/AdminBiz.java b/src/main/java/com/xxl/job/core/biz/AdminBiz.java
new file mode 100644
index 0000000..227a7c4
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/AdminBiz.java
@@ -0,0 +1,42 @@
+package com.xxl.job.core.biz;
+
+import com.xxl.job.core.biz.model.HandleCallbackParam;
+import com.xxl.job.core.biz.model.RegistryParam;
+import com.xxl.job.core.biz.model.ReturnT;
+
+import java.util.List;
+
+/**
+ * @author liyh
+ */
+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);
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java b/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java
new file mode 100644
index 0000000..2d9a9f6
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/ExecutorBiz.java
@@ -0,0 +1,49 @@
+package com.xxl.job.core.biz;
+
+import com.xxl.job.core.biz.model.*;
+
+/**
+ * @author liyh
+ */
+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);
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/client/AdminBizClient.java b/src/main/java/com/xxl/job/core/biz/client/AdminBizClient.java
new file mode 100644
index 0000000..fbcac3a
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/client/AdminBizClient.java
@@ -0,0 +1,51 @@
+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.RegistryParam;
+import com.xxl.job.core.biz.model.ReturnT;
+import com.xxl.job.core.util.XxlJobRemotingUtil;
+
+import java.util.List;
+
+/**
+ * admin api test
+ *
+ * @author liyh
+ */
+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);
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/client/ExecutorBizClient.java b/src/main/java/com/xxl/job/core/biz/client/ExecutorBizClient.java
new file mode 100644
index 0000000..30a24ec
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/client/ExecutorBizClient.java
@@ -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 liyh
+ */
+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);
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java b/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java
new file mode 100644
index 0000000..6f5c22a
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java
@@ -0,0 +1,172 @@
+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;
+
+/**
+ * @author liyh
+ */
+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 old：jobHandler + jobThread
+ JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId());
+ IJobHandler jobHandler = jobThread != null ? jobThread.getHandler() : null;
+ String removeOldReason = null;
+
+ // valid：jobHandler + 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);
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/HandleCallbackParam.java b/src/main/java/com/xxl/job/core/biz/model/HandleCallbackParam.java
new file mode 100644
index 0000000..3fff3d2
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/HandleCallbackParam.java
@@ -0,0 +1,69 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+public class HandleCallbackParam implements Serializable {
+ private static final long serialVersionUID = 42L;
+
+ private long logId;
+ private long logDateTim;
+
+ private int handleCode;
+ private String handleMsg;
+
+ public HandleCallbackParam() {
+ }
+
+ public HandleCallbackParam(long logId, long logDateTim, int handleCode, String handleMsg) {
+ this.logId = logId;
+ this.logDateTim = logDateTim;
+ this.handleCode = handleCode;
+ this.handleMsg = handleMsg;
+ }
+
+ public long getLogId() {
+ return logId;
+ }
+
+ public void setLogId(long 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 + '\'' +
+ '}';
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/IdleBeatParam.java b/src/main/java/com/xxl/job/core/biz/model/IdleBeatParam.java
new file mode 100644
index 0000000..19af462
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/IdleBeatParam.java
@@ -0,0 +1,29 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+public class IdleBeatParam implements Serializable {
+ private static final long serialVersionUID = 42L;
+
+ public IdleBeatParam() {
+ }
+
+ public IdleBeatParam(int jobId) {
+ this.jobId = jobId;
+ }
+
+ private int jobId;
+
+
+ public int getJobId() {
+ return jobId;
+ }
+
+ public void setJobId(int jobId) {
+ this.jobId = jobId;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/KillParam.java b/src/main/java/com/xxl/job/core/biz/model/KillParam.java
new file mode 100644
index 0000000..b455920
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/KillParam.java
@@ -0,0 +1,29 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+public class KillParam implements Serializable {
+ private static final long serialVersionUID = 42L;
+
+ public KillParam() {
+ }
+
+ public KillParam(int jobId) {
+ this.jobId = jobId;
+ }
+
+ private int jobId;
+
+
+ public int getJobId() {
+ return jobId;
+ }
+
+ public void setJobId(int jobId) {
+ this.jobId = jobId;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/LogParam.java b/src/main/java/com/xxl/job/core/biz/model/LogParam.java
new file mode 100644
index 0000000..6bff569
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/LogParam.java
@@ -0,0 +1,48 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+public class LogParam implements Serializable {
+ private static final long serialVersionUID = 42L;
+
+ public LogParam() {
+ }
+
+ public LogParam(long logDateTim, long logId, int fromLineNum) {
+ this.logDateTim = logDateTim;
+ this.logId = logId;
+ this.fromLineNum = fromLineNum;
+ }
+
+ private long logDateTim;
+ private long logId;
+ private int fromLineNum;
+
+ public long getLogDateTim() {
+ return logDateTim;
+ }
+
+ public void setLogDateTim(long logDateTim) {
+ this.logDateTim = logDateTim;
+ }
+
+ public long getLogId() {
+ return logId;
+ }
+
+ public void setLogId(long logId) {
+ this.logId = logId;
+ }
+
+ public int getFromLineNum() {
+ return fromLineNum;
+ }
+
+ public void setFromLineNum(int fromLineNum) {
+ this.fromLineNum = fromLineNum;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/LogResult.java b/src/main/java/com/xxl/job/core/biz/model/LogResult.java
new file mode 100644
index 0000000..2bb658e
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/LogResult.java
@@ -0,0 +1,57 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+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;
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/RegistryParam.java b/src/main/java/com/xxl/job/core/biz/model/RegistryParam.java
new file mode 100644
index 0000000..9c86e4f
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/RegistryParam.java
@@ -0,0 +1,56 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+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 + '\'' +
+ '}';
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/ReturnT.java b/src/main/java/com/xxl/job/core/biz/model/ReturnT.java
new file mode 100644
index 0000000..97ea9a0
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/ReturnT.java
@@ -0,0 +1,65 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * common return
+ *
+ * @author liyh
+ */
+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 + "]";
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/biz/model/TriggerParam.java b/src/main/java/com/xxl/job/core/biz/model/TriggerParam.java
new file mode 100644
index 0000000..67c5028
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/biz/model/TriggerParam.java
@@ -0,0 +1,144 @@
+package com.xxl.job.core.biz.model;
+
+import java.io.Serializable;
+
+/**
+ * @author liyh
+ */
+public class TriggerParam implements Serializable {
+ private static final long serialVersionUID = 42L;
+
+ private int jobId;
+
+ private String executorHandler;
+ private String executorParams;
+ private String executorBlockStrategy;
+ private int executorTimeout;
+
+ private long logId;
+ private long logDateTime;
+
+ private String glueType;
+ private String glueSource;
+ private long glueUpdatetime;
+
+ private int broadcastIndex;
+ private int broadcastTotal;
+
+
+ public int getJobId() {
+ return jobId;
+ }
+
+ public void setJobId(int 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 long getLogId() {
+ return logId;
+ }
+
+ public void setLogId(long 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 +
+ '}';
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/context/XxlJobContext.java b/src/main/java/com/xxl/job/core/context/XxlJobContext.java
new file mode 100644
index 0000000..41a06af
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/context/XxlJobContext.java
@@ -0,0 +1,119 @@
+package com.xxl.job.core.context;
+
+/**
+ * xxl-job context
+ *
+ * @author liyh
+ */
+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 long 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 ----------------------
+
+ /**
+ * handleCode：The result status of job execution
+ * <p>
+ * 200 : success
+ * 500 : fail
+ * 502 : timeout
+ */
+ private int handleCode;
+
+ /**
+ * handleMsg：The simple log msg of job execution
+ */
+ private String handleMsg;
+
+ public XxlJobContext(long 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 long 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();
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/context/XxlJobHelper.java b/src/main/java/com/xxl/job/core/context/XxlJobHelper.java
new file mode 100644
index 0000000..7025ae4
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/context/XxlJobHelper.java
@@ -0,0 +1,252 @@
+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 liyh
+ */
+public class XxlJobHelper {
+
+ // ---------------------- base info ----------------------
+
+ /**
+ * current JobId
+ *
+ * @return
+ */
+ public static long getJobId() {
+ XxlJobContext xxlJobContext = XxlJobContext.getXxlJobContext();
+ if (xxlJobContext == null) {
+ return -1;
+ }
+
+ 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;
+ }
+
+
+}
diff --git a/src/main/java/com/xxl/job/core/enums/ExecutorBlockStrategyEnum.java b/src/main/java/com/xxl/job/core/enums/ExecutorBlockStrategyEnum.java
new file mode 100644
index 0000000..8978d34
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/enums/ExecutorBlockStrategyEnum.java
@@ -0,0 +1,37 @@
+package com.xxl.job.core.enums;
+
+/**
+ * @author liyh
+ */
+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;
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/enums/RegistryConfig.java b/src/main/java/com/xxl/job/core/enums/RegistryConfig.java
new file mode 100644
index 0000000..62533a1
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/enums/RegistryConfig.java
@@ -0,0 +1,13 @@
+package com.xxl.job.core.enums;
+
+/**
+ * @author liyh
+ */
+public class RegistryConfig {
+
+ public static final int BEAT_TIMEOUT = 30;
+ public static final int DEAD_TIMEOUT = BEAT_TIMEOUT * 3;
+
+ public enum RegistType {EXECUTOR, ADMIN}
+
+}
diff --git a/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java b/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java
new file mode 100644
index 0000000..e6695bc
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java
@@ -0,0 +1,279 @@
+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.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;
+
+/**
+ * @author liyh
+ */
+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() {
+ // destroy executor-server
+ stopEmbedServer();
+
+ // destroy jobThreadRepository
+ if (jobThreadRepository.size() > 0) {
+ for (Map.Entry<Integer, 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 {
+
+ // 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-address：default 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<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
+
+ public static JobThread registJobThread(int 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(int jobId, String removeOldReason) {
+ JobThread oldJobThread = jobThreadRepository.remove(jobId);
+ if (oldJobThread != null) {
+ oldJobThread.toStop(removeOldReason);
+ oldJobThread.interrupt();
+
+ return oldJobThread;
+ }
+ return null;
+ }
+
+ public static JobThread loadJobThread(int jobId) {
+ JobThread jobThread = jobThreadRepository.get(jobId);
+ return jobThread;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/executor/impl/XxlJobSimpleExecutor.java b/src/main/java/com/xxl/job/core/executor/impl/XxlJobSimpleExecutor.java
new file mode 100644
index 0000000..428e0a4
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/executor/impl/XxlJobSimpleExecutor.java
@@ -0,0 +1,75 @@
+package com.xxl.job.core.executor.impl;
+
+import com.xxl.job.core.executor.XxlJobExecutor;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
+
+
+/**
+ * xxl-job executor (for frameless)
+ *
+ * @author liyh
+ */
+public class XxlJobSimpleExecutor extends XxlJobExecutor {
+ private static final Logger logger = LoggerFactory.getLogger(XxlJobSimpleExecutor.class);
+
+
+ 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);
+ }
+
+ }
+
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/executor/impl/XxlJobSpringExecutor.java b/src/main/java/com/xxl/job/core/executor/impl/XxlJobSpringExecutor.java
new file mode 100644
index 0000000..292a177
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/executor/impl/XxlJobSpringExecutor.java
@@ -0,0 +1,123 @@
+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.core.MethodIntrospector;
+import org.springframework.core.annotation.AnnotatedElementUtils;
+
+import java.lang.reflect.Method;
+import java.util.Map;
+
+/**
+ * xxl-job executor (for spring)
+ *
+ * @author liyh
+ */
+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);
+ }
+ }
+ }
+ }*/
+
+ 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) {
+ Object bean = applicationContext.getBean(beanDefinitionName);
+
+ 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(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;
+ }
+
+ 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(ApplicationContext applicationContext) throws BeansException {
+ XxlJobSpringExecutor.applicationContext = applicationContext;
+ }
+
+ public static ApplicationContext getApplicationContext() {
+ return applicationContext;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/glue/GlueFactory.java b/src/main/java/com/xxl/job/core/glue/GlueFactory.java
new file mode 100644
index 0000000..9e024b3
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/glue/GlueFactory.java
@@ -0,0 +1,90 @@
+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 liyh
+ */
+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.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) {
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/glue/GlueTypeEnum.java b/src/main/java/com/xxl/job/core/glue/GlueTypeEnum.java
new file mode 100644
index 0000000..be96dc0
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/glue/GlueTypeEnum.java
@@ -0,0 +1,53 @@
+package com.xxl.job.core.glue;
+
+/**
+ * @author liyh
+ */
+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;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/glue/impl/SpringGlueFactory.java b/src/main/java/com/xxl/job/core/glue/impl/SpringGlueFactory.java
new file mode 100644
index 0000000..9d7a5d2
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/glue/impl/SpringGlueFactory.java
@@ -0,0 +1,80 @@
+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 javax.annotation.Resource;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
+
+/**
+ * @author liyh
+ */
+public class SpringGlueFactory extends GlueFactory {
+ private static Logger logger = LoggerFactory.getLogger(SpringGlueFactory.class);
+
+ /**
+ * inject action of spring
+ *
+ * @param instance
+ */
+ @Override
+ 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
+
+ if (AnnotationUtils.getAnnotation(field, Resource.class) != null) {
+ try {
+ Resource resource = AnnotationUtils.getAnnotation(field, Resource.class);
+ if (resource.name() != null && resource.name().length() > 0) {
+ fieldBean = XxlJobSpringExecutor.getApplicationContext().getBean(resource.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);
+ }
+ }
+ }
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/handler/IJobHandler.java b/src/main/java/com/xxl/job/core/handler/IJobHandler.java
new file mode 100644
index 0000000..db520d9
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/IJobHandler.java
@@ -0,0 +1,32 @@
+package com.xxl.job.core.handler;
+
+/**
+ * job handler
+ *
+ * @author liyh
+ */
+public abstract class IJobHandler {
+
+ /**
+ * execute handler, invoked when executor receives a scheduling request
+ *
+ * @throws Exception
+ */
+ public abstract void execute() throws Exception;
+
+ /**
+ * init handler, invoked when JobThread init
+ */
+ public void init() throws Exception {
+
+ }
+
+ /**
+ * destroy handler, invoked when JobThread destroy
+ */
+ public void destroy() throws Exception {
+
+ }
+
+
+}
diff --git a/src/main/java/com/xxl/job/core/handler/annotation/JobHandler.java b/src/main/java/com/xxl/job/core/handler/annotation/JobHandler.java
new file mode 100644
index 0000000..2b560c0
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/annotation/JobHandler.java
@@ -0,0 +1,24 @@
+package com.xxl.job.core.handler.annotation;//package com.xxl.job.core.handler.annotation;//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
+// * <p>
+// * will be replaced by {@link com.xxl.job.core.handler.annotation.XxlJob}
+// *
+// * @author liyh
+// */
+//@Target({ElementType.TYPE})
+//@Retention(RetentionPolicy.RUNTIME)
+//@Inherited
+//@Deprecated
+//public @interface JobHandler {
+//
+// String value();
+//
+//}
diff --git a/src/main/java/com/xxl/job/core/handler/annotation/XxlJob.java b/src/main/java/com/xxl/job/core/handler/annotation/XxlJob.java
new file mode 100644
index 0000000..256f338
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/annotation/XxlJob.java
@@ -0,0 +1,30 @@
+package com.xxl.job.core.handler.annotation;
+
+import java.lang.annotation.*;
+
+/**
+ * annotation for method jobhandler
+ *
+ * @author liyh
+ */
+@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 "";
+
+}
diff --git a/src/main/java/com/xxl/job/core/handler/impl/GlueJobHandler.java b/src/main/java/com/xxl/job/core/handler/impl/GlueJobHandler.java
new file mode 100644
index 0000000..20f99ea
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/impl/GlueJobHandler.java
@@ -0,0 +1,40 @@
+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 liyh
+ */
+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();
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/handler/impl/MethodJobHandler.java b/src/main/java/com/xxl/job/core/handler/impl/MethodJobHandler.java
new file mode 100644
index 0000000..6d3e971
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/impl/MethodJobHandler.java
@@ -0,0 +1,53 @@
+package com.xxl.job.core.handler.impl;
+
+import com.xxl.job.core.handler.IJobHandler;
+
+import java.lang.reflect.Method;
+
+/**
+ * @author liyh
+ */
+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() + "]";
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/handler/impl/ScriptJobHandler.java b/src/main/java/com/xxl/job/core/handler/impl/ScriptJobHandler.java
new file mode 100644
index 0000000..6fc478d
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/handler/impl/ScriptJobHandler.java
@@ -0,0 +1,93 @@
+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;
+
+/**
+ * @author liyh
+ */
+public class ScriptJobHandler extends IJobHandler {
+
+ private int jobId;
+ private long glueUpdatetime;
+ private String gluesource;
+ private GlueTypeEnum glueType;
+
+ public ScriptJobHandler(int 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 params：0=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;
+ }
+
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java b/src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java
new file mode 100644
index 0000000..130085c
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java
@@ -0,0 +1,212 @@
+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 liyh
+ */
+public class XxlJobFileAppender {
+ private static Logger logger = LoggerFactory.getLogger(XxlJobFileAppender.class);
+
+ private static String logBasePath = "logs/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, long 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(String.valueOf(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;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/server/EmbedServer.java b/src/main/java/com/xxl/job/core/server/EmbedServer.java
new file mode 100644
index 0000000..3505fb2
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/server/EmbedServer.java
@@ -0,0 +1,263 @@
+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.*;
+
+/**
+ * @author liyh
+ */
+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) {
+ if (e instanceof InterruptedException) {
+ logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
+ } else {
+ 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
+ *
+ * @author liyh
+ */
+ 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 {
+ if ("/beat".equals(uri)) {
+ return executorBiz.beat();
+ } else if ("/idleBeat".equals(uri)) {
+ IdleBeatParam idleBeatParam = GsonTool.fromJson(requestData, IdleBeatParam.class);
+ return executorBiz.idleBeat(idleBeatParam);
+ } else if ("/run".equals(uri)) {
+ TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);
+ return executorBiz.run(triggerParam);
+ } else if ("/kill".equals(uri)) {
+ KillParam killParam = GsonTool.fromJson(requestData, KillParam.class);
+ return executorBiz.kill(killParam);
+ } else if ("/log".equals(uri)) {
+ LogParam logParam = GsonTool.fromJson(requestData, LogParam.class);
+ return executorBiz.log(logParam);
+ } else {
+ 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();
+ }
+
+
+}
diff --git a/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java b/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java
new file mode 100644
index 0000000..0942911
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/thread/ExecutorRegistryThread.java
@@ -0,0 +1,131 @@
+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;
+
+/**
+ * @author liyh
+ */
+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);
+ }
+ }
+
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/thread/JobLogFileCleanThread.java b/src/main/java/com/xxl/job/core/thread/JobLogFileCleanThread.java
new file mode 100644
index 0000000..c5e537e
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/thread/JobLogFileCleanThread.java
@@ -0,0 +1,126 @@
+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 liyh
+ */
+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);
+ }
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/thread/JobThread.java b/src/main/java/com/xxl/job/core/thread/JobThread.java
new file mode 100644
index 0000000..6070e35
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/thread/JobThread.java
@@ -0,0 +1,255 @@
+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 liyh
+ */
+public class JobThread extends Thread {
+ private static Logger logger = LoggerFactory.getLogger(JobThread.class);
+
+ private int jobId;
+ private IJobHandler handler;
+ private LinkedBlockingQueue<TriggerParam> triggerQueue;
+ private Set<Long> 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(int jobId, IJobHandler handler) {
+ this.jobId = jobId;
+ this.handler = handler;
+ this.triggerQueue = new LinkedBlockingQueue<TriggerParam>();
+ this.triggerLogIdSet = Collections.synchronizedSet(new HashSet<Long>());
+
+ // 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();
+
+ Boolean tempResult = 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 {
+ 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());
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java b/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
new file mode 100644
index 0000000..e46d51b
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/thread/TriggerCallbackThread.java
@@ -0,0 +1,265 @@
+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;
+
+/**
+ * @author liyh
+ */
+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>();
+ int drainToNum = 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>();
+ int drainToNum = 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(
+ -1,
+ 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;
+ }
+
+ List<HandleCallbackParam> callbackParamList = (List<HandleCallbackParam>) JdkSerializeTool.deserialize(callbackParamList_bytes, List.class);
+
+ callbaclLogFile.delete();
+ doCallback(callbackParamList);
+ }
+
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/DateUtil.java b/src/main/java/com/xxl/job/core/util/DateUtil.java
new file mode 100644
index 0000000..4c922fe
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/DateUtil.java
@@ -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 liyh
+ */
+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();
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/FileUtil.java b/src/main/java/com/xxl/job/core/util/FileUtil.java
new file mode 100644
index 0000000..2a38bd7
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/FileUtil.java
@@ -0,0 +1,100 @@
+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 liyh
+ */
+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);
+ }
+ }
+ }
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/GsonTool.java b/src/main/java/com/xxl/job/core/util/GsonTool.java
new file mode 100644
index 0000000..3508408
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/GsonTool.java
@@ -0,0 +1,96 @@
+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 liyh
+ */
+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()
+ );
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/IpUtil.java b/src/main/java/com/xxl/job/core/util/IpUtil.java
new file mode 100644
index 0000000..40955d3
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/IpUtil.java
@@ -0,0 +1,184 @@
+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 liyh
+ */
+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;
+ }
+
+ 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};
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/JdkSerializeTool.java b/src/main/java/com/xxl/job/core/util/JdkSerializeTool.java
new file mode 100644
index 0000000..55647af
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/JdkSerializeTool.java
@@ -0,0 +1,70 @@
+package com.xxl.job.core.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+
+/**
+ * @author liyh
+ */
+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) {
+ ObjectOutputStream oos = null;
+ ByteArrayOutputStream baos = null;
+ try {
+ // 序列化
+ baos = new ByteArrayOutputStream();
+ oos = new ObjectOutputStream(baos);
+ oos.writeObject(object);
+ byte[] bytes = baos.toByteArray();
+ return bytes;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ try {
+ oos.close();
+ baos.close();
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ return null;
+ }
+
+ /**
+ * 将byte[] -->Object
+ *
+ * @param bytes
+ * @return
+ */
+ public static <T> Object deserialize(byte[] bytes, Class<T> clazz) {
+ ByteArrayInputStream bais = null;
+ try {
+ // 反序列化
+ bais = new ByteArrayInputStream(bytes);
+ ObjectInputStream ois = new ObjectInputStream(bais);
+ return ois.readObject();
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ try {
+ bais.close();
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ return null;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/NetUtil.java b/src/main/java/com/xxl/job/core/util/NetUtil.java
new file mode 100644
index 0000000..c77e266
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/NetUtil.java
@@ -0,0 +1,70 @@
+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 liyh
+ */
+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;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/ScriptUtil.java b/src/main/java/com/xxl/job/core/util/ScriptUtil.java
new file mode 100644
index 0000000..9392b71
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/ScriptUtil.java
@@ -0,0 +1,167 @@
+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日志顺序会错乱
+ *
+ * @author liyh
+ */
+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();
+ }
+ }
+ }
+
+ /**
+ * 数据流Copy（Input自动关闭，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);
+ }
+ }
+ }
+ outputStream.flush();
+ //out = null;
+ inputStream.close();
+ inputStream = null;
+ return total;
+ } finally {
+ if (inputStream != null) {
+ inputStream.close();
+ }
+ }
+ }
+}
diff --git a/src/main/java/com/xxl/job/core/util/ThrowableUtil.java b/src/main/java/com/xxl/job/core/util/ThrowableUtil.java
new file mode 100644
index 0000000..011d646
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/ThrowableUtil.java
@@ -0,0 +1,24 @@
+package com.xxl.job.core.util;
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+
+/**
+ * @author liyh
+ */
+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;
+ }
+
+}
diff --git a/src/main/java/com/xxl/job/core/util/XxlJobRemotingUtil.java b/src/main/java/com/xxl/job/core/util/XxlJobRemotingUtil.java
new file mode 100644
index 0000000..0d6974f
--- /dev/null
+++ b/src/main/java/com/xxl/job/core/util/XxlJobRemotingUtil.java
@@ -0,0 +1,158 @@
+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 liyh
+ */
+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(), e);
+ }
+ 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 X509Certificate[] getAcceptedIssuers() {
+ return new X509Certificate[]{};
+ }
+
+ @Override
+ public void checkClientTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+
+ @Override
+ public void checkServerTrusted(X509Certificate[] chain, String authType) throws CertificateException {
+ }
+ }};
+
+ /**
+ * post
+ *
+ * @param url
+ * @param accessToken
+ * @param timeout
+ * @param requestObj
+ * @param returnTargClassOfT
+ * @return
+ */
+ public static ReturnT postBody(String url, String accessToken, int timeout, Object requestObj, Class 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<String>(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 {
+ ReturnT returnT = 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<String>(ReturnT.FAIL_CODE, "xxl-job remoting (url=" + url + ") response content invalid(" + resultJson + ").");
+ }
+
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ return new ReturnT<String>(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(), e2);
+ }
+ }
+ }
+
+}
diff --git a/src/main/java/org/springblade/xxljob/config/XxlJobConfig.java b/src/main/java/org/springblade/xxljob/config/XxlJobConfig.java
new file mode 100644
index 0000000..3f032f2
--- /dev/null
+++ b/src/main/java/org/springblade/xxljob/config/XxlJobConfig.java
@@ -0,0 +1,59 @@
+package org.springblade.xxljob.config;
+
+import com.xxl.job.core.executor.impl.XxlJobSpringExecutor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * xxl-job config
+ *
+ * @author liyh
+ */
+@Configuration
+public class XxlJobConfig {
+ private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
+
+ @Value("${xxl.job.admin.addresses}")
+ private String adminAddresses;
+
+ @Value("${xxl.job.accessToken}")
+ private String accessToken;
+
+ @Value("${xxl.job.executor.appname}")
+ private String appname;
+
+ @Value("${xxl.job.executor.address}")
+ private String address;
+
+ @Value("${xxl.job.executor.ip}")
+ private String ip;
+
+ @Value("${xxl.job.executor.port}")
+ private int port;
+
+ @Value("${xxl.job.executor.logpath}")
+ private String logPath;
+
+ @Value("${xxl.job.executor.logretentiondays}")
+ private int logRetentionDays;
+
+ @Bean
+ public XxlJobSpringExecutor xxlJobExecutor() {
+ logger.info(">>>>>>>>>>> xxl-job config init.");
+ XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
+ xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
+ xxlJobSpringExecutor.setAppname(appname);
+ xxlJobSpringExecutor.setAddress(address);
+ xxlJobSpringExecutor.setIp(ip);
+ xxlJobSpringExecutor.setPort(port);
+ xxlJobSpringExecutor.setAccessToken(accessToken);
+ xxlJobSpringExecutor.setLogPath(logPath);
+ xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
+
+ return xxlJobSpringExecutor;
+ }
+
+}
diff --git a/src/main/java/org/springblade/xxljob/jobhandler/SampleXxlJob.java b/src/main/java/org/springblade/xxljob/jobhandler/SampleXxlJob.java
new file mode 100644
index 0000000..7b28e25
--- /dev/null
+++ b/src/main/java/org/springblade/xxljob/jobhandler/SampleXxlJob.java
@@ -0,0 +1,258 @@
+package org.springblade.xxljob.jobhandler;
+
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Component;
+
+import java.io.BufferedInputStream;
+import java.io.BufferedReader;
+import java.io.DataOutputStream;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * XxlJob开发示例（Bean模式）
+ * <p>
+ * 开发步骤：
+ * 1、任务开发：在Spring Bean实例中，开发Job方法；
+ * 2、注解配置：为Job方法添加注解 "@XxlJob(value="自定义jobhandler名称", init = "JobHandler初始化方法", destroy = "JobHandler销毁方法")"，注解value值对应的是调度中心新建任务的JobHandler属性的值。
+ * 3、执行日志：需要通过 "XxlJobHelper.log" 打印执行日志；
+ * 4、任务结果：默认任务结果为 "成功" 状态，不需要主动设置；如有诉求，比如设置任务结果为失败，可以通过 "XxlJobHelper.handleFail/handleSuccess" 自主设置任务结果；
+ *
+ * @author liyh
+ */
+@Component
+public class SampleXxlJob {
+ private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);
+
+ /**
+ * 1、简单任务示例（Bean模式）
+ */
+ @XxlJob("demoJobHandler")
+ public void demoJobHandler() throws Exception {
+ XxlJobHelper.log("XXL-JOB, Hello World.");
+
+ for (int i = 0; i < 5; i++) {
+ XxlJobHelper.log("beat at:" + i);
+ System.out.println("hello" + i);
+ TimeUnit.SECONDS.sleep(2);
+ }
+ System.out.println("完成了");
+ }
+
+ @XxlJob("testJobHandler")
+ public void testJobHandler() throws Exception {
+ String jobParam = XxlJobHelper.getJobParam();
+ System.out.println("执行成功，参数：" + jobParam);
+ }
+
+ /**
+ * 2、分片广播任务
+ */
+ @XxlJob("shardingJobHandler")
+ public void shardingJobHandler() throws Exception {
+
+ // 分片参数
+ int shardIndex = XxlJobHelper.getShardIndex();
+ int shardTotal = XxlJobHelper.getShardTotal();
+
+ XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
+
+ // 业务逻辑
+ for (int i = 0; i < shardTotal; i++) {
+ if (i == shardIndex) {
+ XxlJobHelper.log("第 {} 片, 命中分片开始处理", i);
+ } else {
+ XxlJobHelper.log("第 {} 片, 忽略", i);
+ }
+ }
+
+ }
+
+ /**
+ * 3、命令行任务
+ */
+ @XxlJob("commandJobHandler")
+ public void commandJobHandler() throws Exception {
+ String command = XxlJobHelper.getJobParam();
+ int exitValue = -1;
+
+ BufferedReader bufferedReader = null;
+ try {
+ // command process
+ ProcessBuilder processBuilder = new ProcessBuilder();
+ processBuilder.command(command);
+ processBuilder.redirectErrorStream(true);
+
+ Process process = processBuilder.start();
+ //Process process = Runtime.getRuntime().exec(command);
+
+ BufferedInputStream bufferedInputStream = new BufferedInputStream(process.getInputStream());
+ bufferedReader = new BufferedReader(new InputStreamReader(bufferedInputStream));
+
+ // command log
+ String line;
+ while ((line = bufferedReader.readLine()) != null) {
+ XxlJobHelper.log(line);
+ }
+
+ // command exit
+ process.waitFor();
+ exitValue = process.exitValue();
+ } catch (Exception e) {
+ XxlJobHelper.log(e);
+ } finally {
+ if (bufferedReader != null) {
+ bufferedReader.close();
+ }
+ }
+
+ if (exitValue == 0) {
+ // default success
+ } else {
+ XxlJobHelper.handleFail("command exit value(" + exitValue + ") is failed");
+ }
+
+ }
+
+
+ /**
+ * 4、跨平台Http任务
+ * 参数示例：
+ * "url: http://www.baidu.com\n" +
+ * "method: get\n" +
+ * "data: content\n";
+ */
+ @XxlJob("httpJobHandler")
+ public void httpJobHandler() throws Exception {
+
+ // param parse
+ String param = XxlJobHelper.getJobParam();
+ if (param == null || param.trim().length() == 0) {
+ XxlJobHelper.log("param[" + param + "] invalid.");
+
+ XxlJobHelper.handleFail();
+ return;
+ }
+
+ String[] httpParams = param.split("\n");
+ String url = null;
+ String method = null;
+ String data = null;
+ for (String httpParam : httpParams) {
+ if (httpParam.startsWith("url:")) {
+ url = httpParam.substring(httpParam.indexOf("url:") + 4).trim();
+ }
+ if (httpParam.startsWith("method:")) {
+ method = httpParam.substring(httpParam.indexOf("method:") + 7).trim().toUpperCase();
+ }
+ if (httpParam.startsWith("data:")) {
+ data = httpParam.substring(httpParam.indexOf("data:") + 5).trim();
+ }
+ }
+
+ // param valid
+ if (url == null || url.trim().length() == 0) {
+ XxlJobHelper.log("url[" + url + "] invalid.");
+
+ XxlJobHelper.handleFail();
+ return;
+ }
+ if (method == null || !Arrays.asList("GET", "POST").contains(method)) {
+ XxlJobHelper.log("method[" + method + "] invalid.");
+
+ XxlJobHelper.handleFail();
+ return;
+ }
+ boolean isPostMethod = method.equals("POST");
+
+ // request
+ HttpURLConnection connection = null;
+ BufferedReader bufferedReader = null;
+ try {
+ // connection
+ URL realUrl = new URL(url);
+ connection = (HttpURLConnection) realUrl.openConnection();
+
+ // connection setting
+ connection.setRequestMethod(method);
+ connection.setDoOutput(isPostMethod);
+ connection.setDoInput(true);
+ connection.setUseCaches(false);
+ connection.setReadTimeout(5 * 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");
+
+ // do connection
+ connection.connect();
+
+ // data
+ if (isPostMethod && data != null && data.trim().length() > 0) {
+ DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
+ dataOutputStream.write(data.getBytes("UTF-8"));
+ dataOutputStream.flush();
+ dataOutputStream.close();
+ }
+
+ // valid StatusCode
+ int statusCode = connection.getResponseCode();
+ if (statusCode != 200) {
+ throw new RuntimeException("Http Request StatusCode(" + statusCode + ") Invalid.");
+ }
+
+ // 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 responseMsg = result.toString();
+
+ XxlJobHelper.log(responseMsg);
+
+ return;
+ } catch (Exception e) {
+ XxlJobHelper.log(e);
+
+ XxlJobHelper.handleFail();
+ return;
+ } finally {
+ try {
+ if (bufferedReader != null) {
+ bufferedReader.close();
+ }
+ if (connection != null) {
+ connection.disconnect();
+ }
+ } catch (Exception e2) {
+ XxlJobHelper.log(e2);
+ }
+ }
+
+ }
+
+ /**
+ * 5、生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑；
+ */
+ @XxlJob(value = "demoJobHandler2", init = "init", destroy = "destroy")
+ public void demoJobHandler2() throws Exception {
+ XxlJobHelper.log("XXL-JOB, Hello World.");
+ }
+
+ public void init() {
+ logger.info("init");
+ }
+
+ public void destroy() {
+ logger.info("destroy");
+ }
+
+}
diff --git a/src/main/resources/application-dev.yml b/src/main/resources/application-dev.yml
index 47c9307..6550629 100644
--- a/src/main/resources/application-dev.yml
+++ b/src/main/resources/application-dev.yml
@@ -52,3 +52,16 @@
upload-domain: http://localhost:8999
remote-path: /usr/share/nginx/html
+# xxl-job
+xxl:
+ job:
+ accessToken: ''
+ admin:
+ addresses: http://192.168.1.50:7009/xxl-job-admin
+ executor:
+ appname: blade-xxljob
+ ip: 127.0.0.1
+ logpath: ../data/applogs/xxl-job/jobhandler
+ logretentiondays: -1
+ port: 7018
+ address:
diff --git a/src/main/resources/application-prod.yml b/src/main/resources/application-prod.yml
index de331c1..3d8d416 100644
--- a/src/main/resources/application-prod.yml
+++ b/src/main/resources/application-prod.yml
@@ -38,3 +38,17 @@
remote-mode: true
upload-domain: http://localhost:8999
remote-path: /usr/share/nginx/html
+
+# xxl-job
+xxl:
+ job:
+ accessToken: ''
+ admin:
+ addresses: http://127.0.0.1:7009/xxl-job-admin
+ executor:
+ appname: blade-xxljob
+ ip: 127.0.0.1
+ logpath: /app/server/xxl-job/jobhandler
+ logretentiondays: -1
+ port: 7018
+ address:
diff --git a/src/main/resources/application-test.yml b/src/main/resources/application-test.yml
index e3fe180..6e80943 100644
--- a/src/main/resources/application-test.yml
+++ b/src/main/resources/application-test.yml
@@ -38,3 +38,17 @@
remote-mode: true
upload-domain: http://localhost:8999
remote-path: /usr/share/nginx/html
+
+# xxl-job
+xxl:
+ job:
+ accessToken: ''
+ admin:
+ addresses: http://192.168.1.50:7009/xxl-job-admin
+ executor:
+ appname: blade-xxljob
+ ip: 127.0.0.1
+ logpath: ../data/applogs/xxl-job/jobhandler
+ logretentiondays: -1
+ port: 7018
+ address:
--
Gitblit v1.9.3