|
@@ -12,6 +12,7 @@ import com.xxl.job.core.thread.JobThread;
|
|
|
import com.xxl.job.core.thread.TriggerCallbackThread;
|
|
import com.xxl.job.core.thread.TriggerCallbackThread;
|
|
|
import com.xxl.job.core.util.IpUtil;
|
|
import com.xxl.job.core.util.IpUtil;
|
|
|
import com.xxl.job.core.util.NetUtil;
|
|
import com.xxl.job.core.util.NetUtil;
|
|
|
|
|
+import org.apache.groovy.parser.antlr4.util.StringUtils;
|
|
|
import org.slf4j.Logger;
|
|
import org.slf4j.Logger;
|
|
|
import org.slf4j.LoggerFactory;
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
|
|
@@ -25,7 +26,7 @@ import java.util.concurrent.ConcurrentMap;
|
|
|
/**
|
|
/**
|
|
|
* Created by xuxueli on 2016/3/2 21:14.
|
|
* Created by xuxueli on 2016/3/2 21:14.
|
|
|
*/
|
|
*/
|
|
|
-public class XxlJobExecutor {
|
|
|
|
|
|
|
+public class XxlJobExecutor {
|
|
|
private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
|
|
private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
|
|
|
|
|
|
|
|
// ---------------------- param ----------------------
|
|
// ---------------------- param ----------------------
|
|
@@ -34,6 +35,7 @@ public class XxlJobExecutor {
|
|
|
private String appname;
|
|
private String appname;
|
|
|
private String address;
|
|
private String address;
|
|
|
private String ip;
|
|
private String ip;
|
|
|
|
|
+ private String includeIpPrefix;
|
|
|
private int port;
|
|
private int port;
|
|
|
private String logPath;
|
|
private String logPath;
|
|
|
private int logRetentionDays;
|
|
private int logRetentionDays;
|
|
@@ -41,24 +43,35 @@ public class XxlJobExecutor {
|
|
|
public void setAdminAddresses(String adminAddresses) {
|
|
public void setAdminAddresses(String adminAddresses) {
|
|
|
this.adminAddresses = adminAddresses;
|
|
this.adminAddresses = adminAddresses;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setAccessToken(String accessToken) {
|
|
public void setAccessToken(String accessToken) {
|
|
|
this.accessToken = accessToken;
|
|
this.accessToken = accessToken;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setAppname(String appname) {
|
|
public void setAppname(String appname) {
|
|
|
this.appname = appname;
|
|
this.appname = appname;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setAddress(String address) {
|
|
public void setAddress(String address) {
|
|
|
this.address = address;
|
|
this.address = address;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setIp(String ip) {
|
|
public void setIp(String ip) {
|
|
|
this.ip = ip;
|
|
this.ip = ip;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
|
|
+ public void setIncludeIpPrefix(String includeIpPrefix) {
|
|
|
|
|
+ this.includeIpPrefix = includeIpPrefix;
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
public void setPort(int port) {
|
|
public void setPort(int port) {
|
|
|
this.port = port;
|
|
this.port = port;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setLogPath(String logPath) {
|
|
public void setLogPath(String logPath) {
|
|
|
this.logPath = logPath;
|
|
this.logPath = logPath;
|
|
|
}
|
|
}
|
|
|
|
|
+
|
|
|
public void setLogRetentionDays(int logRetentionDays) {
|
|
public void setLogRetentionDays(int logRetentionDays) {
|
|
|
this.logRetentionDays = logRetentionDays;
|
|
this.logRetentionDays = logRetentionDays;
|
|
|
}
|
|
}
|
|
@@ -66,7 +79,6 @@ public class XxlJobExecutor {
|
|
|
|
|
|
|
|
// ---------------------- start + stop ----------------------
|
|
// ---------------------- start + stop ----------------------
|
|
|
public void start() throws Exception {
|
|
public void start() throws Exception {
|
|
|
-
|
|
|
|
|
// init logpath
|
|
// init logpath
|
|
|
XxlJobFileAppender.initLogPath(logPath);
|
|
XxlJobFileAppender.initLogPath(logPath);
|
|
|
|
|
|
|
@@ -83,13 +95,19 @@ public class XxlJobExecutor {
|
|
|
// init executor-server
|
|
// init executor-server
|
|
|
initEmbedServer(address, ip, port, appname, accessToken);
|
|
initEmbedServer(address, ip, port, appname, accessToken);
|
|
|
}
|
|
}
|
|
|
- public void destroy(){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public boolean checkIncludeIpPrefix() {
|
|
|
|
|
+ String realIp = (ip != null && !ip.trim().isEmpty()) ? ip : IpUtil.getIp();
|
|
|
|
|
+ return includeIpPrefix == null || includeIpPrefix.trim().isEmpty() || includeIpPrefix.contains(realIp.split("\\.")[0]);
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
|
|
+ public void destroy() {
|
|
|
// destroy executor-server
|
|
// destroy executor-server
|
|
|
stopEmbedServer();
|
|
stopEmbedServer();
|
|
|
|
|
|
|
|
// destroy jobThreadRepository
|
|
// destroy jobThreadRepository
|
|
|
if (jobThreadRepository.size() > 0) {
|
|
if (jobThreadRepository.size() > 0) {
|
|
|
- for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) {
|
|
|
|
|
|
|
+ for (Map.Entry<Integer, JobThread> item : jobThreadRepository.entrySet()) {
|
|
|
JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
|
|
JobThread oldJobThread = removeJobThread(item.getKey(), "web container destroy and kill the job.");
|
|
|
// wait for job thread push result to callback queue
|
|
// wait for job thread push result to callback queue
|
|
|
if (oldJobThread != null) {
|
|
if (oldJobThread != null) {
|
|
@@ -116,10 +134,11 @@ public class XxlJobExecutor {
|
|
|
|
|
|
|
|
// ---------------------- admin-client (rpc invoker) ----------------------
|
|
// ---------------------- admin-client (rpc invoker) ----------------------
|
|
|
private static List<AdminBiz> adminBizList;
|
|
private static List<AdminBiz> adminBizList;
|
|
|
|
|
+
|
|
|
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
|
|
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) {
|
|
|
|
|
|
|
+ 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);
|
|
AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
|
|
|
|
|
|
|
@@ -131,7 +150,8 @@ public class XxlJobExecutor {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
- public static List<AdminBiz> getAdminBizList(){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static List<AdminBiz> getAdminBizList() {
|
|
|
return adminBizList;
|
|
return adminBizList;
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -141,17 +161,17 @@ public class XxlJobExecutor {
|
|
|
private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
|
|
private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
|
|
|
|
|
|
|
|
// fill ip port
|
|
// fill ip port
|
|
|
- port = port>0?port: NetUtil.findAvailablePort(9999);
|
|
|
|
|
- ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
|
|
|
|
|
|
|
+ port = port > 0 ? port : NetUtil.findAvailablePort(9999);
|
|
|
|
|
+ ip = (ip != null && ip.trim().length() > 0) ? ip : IpUtil.getIp();
|
|
|
|
|
|
|
|
// generate address
|
|
// generate address
|
|
|
- if (address==null || address.trim().length()==0) {
|
|
|
|
|
|
|
+ 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
|
|
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);
|
|
address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
// accessToken
|
|
// accessToken
|
|
|
- if (accessToken==null || accessToken.trim().length()==0) {
|
|
|
|
|
|
|
+ if (accessToken == null || accessToken.trim().length() == 0) {
|
|
|
logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
|
|
logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -174,14 +194,17 @@ public class XxlJobExecutor {
|
|
|
|
|
|
|
|
// ---------------------- job handler repository ----------------------
|
|
// ---------------------- job handler repository ----------------------
|
|
|
private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
|
|
private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
|
|
|
- public static IJobHandler loadJobHandler(String name){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static IJobHandler loadJobHandler(String name) {
|
|
|
return jobHandlerRepository.get(name);
|
|
return jobHandlerRepository.get(name);
|
|
|
}
|
|
}
|
|
|
- public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static IJobHandler registJobHandler(String name, IJobHandler jobHandler) {
|
|
|
logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
|
|
logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
|
|
|
return jobHandlerRepository.put(name, jobHandler);
|
|
return jobHandlerRepository.put(name, jobHandler);
|
|
|
}
|
|
}
|
|
|
- protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod) {
|
|
|
if (xxlJob == null) {
|
|
if (xxlJob == null) {
|
|
|
return;
|
|
return;
|
|
|
}
|
|
}
|
|
@@ -238,12 +261,13 @@ public class XxlJobExecutor {
|
|
|
|
|
|
|
|
// ---------------------- job thread repository ----------------------
|
|
// ---------------------- job thread repository ----------------------
|
|
|
private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
|
|
private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
|
|
|
- public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason) {
|
|
|
JobThread newJobThread = new JobThread(jobId, handler);
|
|
JobThread newJobThread = new JobThread(jobId, handler);
|
|
|
newJobThread.start();
|
|
newJobThread.start();
|
|
|
logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});
|
|
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!!!
|
|
|
|
|
|
|
+ JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); // putIfAbsent | oh my god, map's put method return the old value!!!
|
|
|
if (oldJobThread != null) {
|
|
if (oldJobThread != null) {
|
|
|
oldJobThread.toStop(removeOldReason);
|
|
oldJobThread.toStop(removeOldReason);
|
|
|
oldJobThread.interrupt();
|
|
oldJobThread.interrupt();
|
|
@@ -251,7 +275,8 @@ public class XxlJobExecutor {
|
|
|
|
|
|
|
|
return newJobThread;
|
|
return newJobThread;
|
|
|
}
|
|
}
|
|
|
- public static JobThread removeJobThread(int jobId, String removeOldReason){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static JobThread removeJobThread(int jobId, String removeOldReason) {
|
|
|
JobThread oldJobThread = jobThreadRepository.remove(jobId);
|
|
JobThread oldJobThread = jobThreadRepository.remove(jobId);
|
|
|
if (oldJobThread != null) {
|
|
if (oldJobThread != null) {
|
|
|
oldJobThread.toStop(removeOldReason);
|
|
oldJobThread.toStop(removeOldReason);
|
|
@@ -261,7 +286,8 @@ public class XxlJobExecutor {
|
|
|
}
|
|
}
|
|
|
return null;
|
|
return null;
|
|
|
}
|
|
}
|
|
|
- public static JobThread loadJobThread(int jobId){
|
|
|
|
|
|
|
+
|
|
|
|
|
+ public static JobThread loadJobThread(int jobId) {
|
|
|
JobThread jobThread = jobThreadRepository.get(jobId);
|
|
JobThread jobThread = jobThreadRepository.get(jobId);
|
|
|
return jobThread;
|
|
return jobThread;
|
|
|
}
|
|
}
|