|
@@ -1,26 +1,21 @@
|
|
|
package com.xxl.job.core.executor;
|
|
|
|
|
|
import com.xxl.job.core.biz.AdminBiz;
|
|
|
-import com.xxl.job.core.biz.ExecutorBiz;
|
|
|
import com.xxl.job.core.biz.client.AdminBizClient;
|
|
|
-import com.xxl.job.core.biz.impl.ExecutorBizImpl;
|
|
|
import com.xxl.job.core.handler.IJobHandler;
|
|
|
import com.xxl.job.core.log.XxlJobFileAppender;
|
|
|
-import com.xxl.job.core.thread.ExecutorRegistryThread;
|
|
|
+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.rpc.registry.ServiceRegistry;
|
|
|
-import com.xxl.rpc.remoting.net.impl.netty_http.server.NettyHttpServer;
|
|
|
-import com.xxl.rpc.remoting.provider.XxlRpcProviderFactory;
|
|
|
-import com.xxl.rpc.serialize.Serializer;
|
|
|
-import com.xxl.rpc.serialize.impl.HessianSerializer;
|
|
|
-import com.xxl.rpc.util.IpUtil;
|
|
|
-import com.xxl.rpc.util.NetUtil;
|
|
|
+import com.xxl.job.core.util.IpUtil;
|
|
|
+import com.xxl.job.core.util.NetUtil;
|
|
|
import org.slf4j.Logger;
|
|
|
import org.slf4j.LoggerFactory;
|
|
|
|
|
|
-import java.util.*;
|
|
|
+import java.util.ArrayList;
|
|
|
+import java.util.List;
|
|
|
+import java.util.Map;
|
|
|
import java.util.concurrent.ConcurrentHashMap;
|
|
|
import java.util.concurrent.ConcurrentMap;
|
|
|
|
|
@@ -32,18 +27,25 @@ public class XxlJobExecutor {
|
|
|
|
|
|
// ---------------------- param ----------------------
|
|
|
private String adminAddresses;
|
|
|
- private String appName;
|
|
|
+ private String accessToken;
|
|
|
+ private String appname;
|
|
|
+ private String address;
|
|
|
private String ip;
|
|
|
private int port;
|
|
|
- private String accessToken;
|
|
|
private String logPath;
|
|
|
private int logRetentionDays;
|
|
|
|
|
|
public void setAdminAddresses(String adminAddresses) {
|
|
|
this.adminAddresses = adminAddresses;
|
|
|
}
|
|
|
- public void setAppName(String appName) {
|
|
|
- this.appName = appName;
|
|
|
+ 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;
|
|
@@ -51,9 +53,6 @@ public class XxlJobExecutor {
|
|
|
public void setPort(int port) {
|
|
|
this.port = port;
|
|
|
}
|
|
|
- public void setAccessToken(String accessToken) {
|
|
|
- this.accessToken = accessToken;
|
|
|
- }
|
|
|
public void setLogPath(String logPath) {
|
|
|
this.logPath = logPath;
|
|
|
}
|
|
@@ -79,18 +78,24 @@ public class XxlJobExecutor {
|
|
|
TriggerCallbackThread.getInstance().start();
|
|
|
|
|
|
// init executor-server
|
|
|
- port = port>0?port: NetUtil.findAvailablePort(9999);
|
|
|
- ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
|
|
|
- initRpcProvider(ip, port, appName, accessToken);
|
|
|
+ initEmbedServer(address, ip, port, appname, accessToken);
|
|
|
}
|
|
|
public void destroy(){
|
|
|
// destory executor-server
|
|
|
- stopRpcProvider();
|
|
|
+ stopEmbedServer();
|
|
|
|
|
|
// destory jobThreadRepository
|
|
|
if (jobThreadRepository.size() > 0) {
|
|
|
for (Map.Entry<Integer, JobThread> item: jobThreadRepository.entrySet()) {
|
|
|
- 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
|
|
|
+ if (oldJobThread != null) {
|
|
|
+ try {
|
|
|
+ oldJobThread.join();
|
|
|
+ } catch (InterruptedException e) {
|
|
|
+ logger.error(">>>>>>>>>>> xxl-job, JobThread destroy(join) error, jobId:{}", item.getKey(), e);
|
|
|
+ }
|
|
|
+ }
|
|
|
}
|
|
|
jobThreadRepository.clear();
|
|
|
}
|
|
@@ -108,7 +113,6 @@ public class XxlJobExecutor {
|
|
|
|
|
|
// ---------------------- admin-client (rpc invoker) ----------------------
|
|
|
private static List<AdminBiz> adminBizList;
|
|
|
- private static Serializer serializer = new HessianSerializer();
|
|
|
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
|
|
|
if (adminAddresses!=null && adminAddresses.trim().length()>0) {
|
|
|
for (String address: adminAddresses.trim().split(",")) {
|
|
@@ -127,78 +131,36 @@ public class XxlJobExecutor {
|
|
|
public static List<AdminBiz> getAdminBizList(){
|
|
|
return adminBizList;
|
|
|
}
|
|
|
- public static Serializer getSerializer() {
|
|
|
- return serializer;
|
|
|
- }
|
|
|
-
|
|
|
|
|
|
// ---------------------- executor-server (rpc provider) ----------------------
|
|
|
- private XxlRpcProviderFactory xxlRpcProviderFactory = null;
|
|
|
-
|
|
|
- private void initRpcProvider(String ip, int port, String appName, String accessToken) throws Exception {
|
|
|
-
|
|
|
- // init, provider factory
|
|
|
- String address = IpUtil.getIpPort(ip, port);
|
|
|
- Map<String, String> serviceRegistryParam = new HashMap<String, String>();
|
|
|
- serviceRegistryParam.put("appName", appName);
|
|
|
- serviceRegistryParam.put("address", address);
|
|
|
-
|
|
|
- xxlRpcProviderFactory = new XxlRpcProviderFactory();
|
|
|
-
|
|
|
- xxlRpcProviderFactory.setServer(NettyHttpServer.class);
|
|
|
- xxlRpcProviderFactory.setSerializer(HessianSerializer.class);
|
|
|
- xxlRpcProviderFactory.setCorePoolSize(20);
|
|
|
- xxlRpcProviderFactory.setMaxPoolSize(200);
|
|
|
- xxlRpcProviderFactory.setIp(ip);
|
|
|
- xxlRpcProviderFactory.setPort(port);
|
|
|
- xxlRpcProviderFactory.setAccessToken(accessToken);
|
|
|
- xxlRpcProviderFactory.setServiceRegistry(ExecutorServiceRegistry.class);
|
|
|
- xxlRpcProviderFactory.setServiceRegistryParam(serviceRegistryParam);
|
|
|
+ private EmbedServer embedServer = null;
|
|
|
|
|
|
- // add services
|
|
|
- xxlRpcProviderFactory.addService(ExecutorBiz.class.getName(), null, new ExecutorBizImpl());
|
|
|
+ private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
|
|
|
|
|
|
- // start
|
|
|
- xxlRpcProviderFactory.start();
|
|
|
-
|
|
|
- }
|
|
|
-
|
|
|
- public static class ExecutorServiceRegistry extends ServiceRegistry {
|
|
|
+ // fill ip port
|
|
|
+ port = port>0?port: NetUtil.findAvailablePort(9999);
|
|
|
+ ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
|
|
|
|
|
|
- @Override
|
|
|
- public void start(Map<String, String> param) {
|
|
|
- // start registry
|
|
|
- ExecutorRegistryThread.getInstance().start(param.get("appName"), param.get("address"));
|
|
|
- }
|
|
|
- @Override
|
|
|
- public void stop() {
|
|
|
- // stop registry
|
|
|
- ExecutorRegistryThread.getInstance().toStop();
|
|
|
+ // 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);
|
|
|
}
|
|
|
|
|
|
- @Override
|
|
|
- public boolean registry(Set<String> keys, String value) {
|
|
|
- return false;
|
|
|
- }
|
|
|
- @Override
|
|
|
- public boolean remove(Set<String> keys, String value) {
|
|
|
- return false;
|
|
|
- }
|
|
|
- @Override
|
|
|
- public Map<String, TreeSet<String>> discovery(Set<String> keys) {
|
|
|
- return null;
|
|
|
- }
|
|
|
- @Override
|
|
|
- public TreeSet<String> discovery(String key) {
|
|
|
- return null;
|
|
|
+ // 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 stopRpcProvider() {
|
|
|
+ private void stopEmbedServer() {
|
|
|
// stop provider factory
|
|
|
try {
|
|
|
- xxlRpcProviderFactory.stop();
|
|
|
+ embedServer.stop();
|
|
|
} catch (Exception e) {
|
|
|
logger.error(e.getMessage(), e);
|
|
|
}
|
|
@@ -207,13 +169,13 @@ public class XxlJobExecutor {
|
|
|
|
|
|
// ---------------------- 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);
|
|
|
}
|
|
|
- public static IJobHandler loadJobHandler(String name){
|
|
|
- return jobHandlerRepository.get(name);
|
|
|
- }
|
|
|
|
|
|
|
|
|
// ---------------------- job thread repository ----------------------
|
|
@@ -231,12 +193,15 @@ public class XxlJobExecutor {
|
|
|
|
|
|
return newJobThread;
|
|
|
}
|
|
|
- public static void removeJobThread(int jobId, String removeOldReason){
|
|
|
+ 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);
|