|  | @@ -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);
 |