Explorar el Código

执行器参数配置逻辑重构

xuxueli hace 8 años
padre
commit
09d8952448

+ 63 - 61
xxl-job-core/src/main/java/com/xxl/job/core/executor/XxlJobExecutor.java Ver fichero

5
 import com.xxl.job.core.biz.impl.ExecutorBizImpl;
5
 import com.xxl.job.core.biz.impl.ExecutorBizImpl;
6
 import com.xxl.job.core.handler.IJobHandler;
6
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.handler.annotation.JobHander;
7
 import com.xxl.job.core.handler.annotation.JobHander;
8
+import com.xxl.job.core.log.XxlJobFileAppender;
8
 import com.xxl.job.core.rpc.netcom.NetComClientProxy;
9
 import com.xxl.job.core.rpc.netcom.NetComClientProxy;
9
 import com.xxl.job.core.rpc.netcom.NetComServerFactory;
10
 import com.xxl.job.core.rpc.netcom.NetComServerFactory;
10
-import com.xxl.job.core.thread.ExecutorRegistryThread;
11
 import com.xxl.job.core.thread.JobThread;
11
 import com.xxl.job.core.thread.JobThread;
12
-import com.xxl.job.core.thread.TriggerCallbackThread;
13
 import org.slf4j.Logger;
12
 import org.slf4j.Logger;
14
 import org.slf4j.LoggerFactory;
13
 import org.slf4j.LoggerFactory;
15
 import org.springframework.beans.BeansException;
14
 import org.springframework.beans.BeansException;
16
 import org.springframework.context.ApplicationContext;
15
 import org.springframework.context.ApplicationContext;
17
 import org.springframework.context.ApplicationContextAware;
16
 import org.springframework.context.ApplicationContextAware;
18
-import org.springframework.context.ApplicationEvent;
19
-import org.springframework.context.ApplicationListener;
20
-import org.springframework.context.event.ContextClosedEvent;
21
 
17
 
22
 import java.util.ArrayList;
18
 import java.util.ArrayList;
23
 import java.util.List;
19
 import java.util.List;
27
 /**
23
 /**
28
  * Created by xuxueli on 2016/3/2 21:14.
24
  * Created by xuxueli on 2016/3/2 21:14.
29
  */
25
  */
30
-public class XxlJobExecutor implements ApplicationContextAware, ApplicationListener {
26
+public class XxlJobExecutor implements ApplicationContextAware {
31
     private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
27
     private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);
32
 
28
 
29
+    // ---------------------------------- param ------------------------------------
33
     private String ip;
30
     private String ip;
34
     private int port = 9999;
31
     private int port = 9999;
35
     private String appName;
32
     private String appName;
36
     private String adminAddresses;
33
     private String adminAddresses;
37
     private String accessToken;
34
     private String accessToken;
38
-    public static String logPath = "/data/applogs/xxl-job/jobhandler/";
35
+    private String logPath;
39
 
36
 
40
     public void setIp(String ip) {
37
     public void setIp(String ip) {
41
         this.ip = ip;
38
         this.ip = ip;
56
         this.logPath = logPath;
53
         this.logPath = logPath;
57
     }
54
     }
58
 
55
 
56
+
57
+    // ---------------------------------- applicationContext ------------------------------------
58
+    private static ApplicationContext applicationContext;
59
+    @Override
60
+    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
61
+        this.applicationContext = applicationContext;
62
+    }
63
+    public static ApplicationContext getApplicationContext() {
64
+        return applicationContext;
65
+    }
66
+
67
+
68
+    // ---------------------------------- start + stop ------------------------------------
69
+    public void start() throws Exception {
70
+        // init admin-client
71
+        initAdminBizList(adminAddresses, accessToken);
72
+
73
+        // init executor-jobHandlerRepository
74
+        initJobHandlerRepository(applicationContext);
75
+
76
+        // init logpath
77
+        if (logPath!=null && logPath.trim().length()>0) {
78
+            XxlJobFileAppender.logPath = logPath;
79
+        }
80
+
81
+        // init executor-server
82
+        initExecutorServer();
83
+    }
84
+    public void destroy(){
85
+        // destory JobThreadRepository
86
+        if (JobThreadRepository.size() > 0) {
87
+            for (Map.Entry<Integer, JobThread> item: JobThreadRepository.entrySet()) {
88
+                removeJobThread(item.getKey(), "Web容器销毁终止");
89
+            }
90
+            JobThreadRepository.clear();
91
+        }
92
+
93
+        // destory executor-server
94
+        stopExecutorServer();
95
+    }
96
+
97
+
59
     // ---------------------------------- admin-client ------------------------------------
98
     // ---------------------------------- admin-client ------------------------------------
60
     private static List<AdminBiz> adminBizList;
99
     private static List<AdminBiz> adminBizList;
61
     private static void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
100
     private static void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
76
         return adminBizList;
115
         return adminBizList;
77
     }
116
     }
78
 
117
 
79
-    // ---------------------------------- job server ------------------------------------
80
-    private NetComServerFactory serverFactory = new NetComServerFactory();
81
-    public void start() throws Exception {
82
-        // init admin-client
83
-        initAdminBizList(adminAddresses, accessToken);
84
 
118
 
85
-        // executor start
119
+    // ---------------------------------- executor-server ------------------------------------
120
+    private NetComServerFactory serverFactory = new NetComServerFactory();
121
+    private void initExecutorServer() throws Exception {
86
         NetComServerFactory.putService(ExecutorBiz.class, new ExecutorBizImpl());   // rpc-service, base on jetty
122
         NetComServerFactory.putService(ExecutorBiz.class, new ExecutorBizImpl());   // rpc-service, base on jetty
87
         NetComServerFactory.setAccessToken(accessToken);
123
         NetComServerFactory.setAccessToken(accessToken);
88
-        serverFactory.start(port, ip, appName);
89
-
90
-
91
-        // trigger callback thread start
92
-        TriggerCallbackThread.getInstance().start();
124
+        serverFactory.start(port, ip, appName); // jetty + registry
93
     }
125
     }
94
-    public void destroy(){
95
-        // 1、executor registry thread stop
96
-        ExecutorRegistryThread.getInstance().toStop();
97
-
98
-        // 2、executor stop
99
-        serverFactory.destroy();
100
-
101
-        // 3、job thread repository destory
102
-        if (JobThreadRepository.size() > 0) {
103
-            for (Map.Entry<Integer, JobThread> item: JobThreadRepository.entrySet()) {
104
-                JobThread jobThread = item.getValue();
105
-                jobThread.toStop("Web容器销毁终止");
106
-                jobThread.interrupt();
107
-
108
-            }
109
-            JobThreadRepository.clear();
110
-        }
111
-
112
-        // 4、trigger callback thread stop
113
-        TriggerCallbackThread.getInstance().toStop();
126
+    private void stopExecutorServer() {
127
+        serverFactory.destroy();    // jetty + registry + callback
114
     }
128
     }
115
 
129
 
116
-    // ---------------------------------- init job handler ------------------------------------
117
-    public static ApplicationContext applicationContext;
118
-	@Override
119
-	public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
120
-        XxlJobExecutor.applicationContext = applicationContext;
121
 
130
 
131
+    // ---------------------------------- job handler repository ------------------------------------
132
+    private static ConcurrentHashMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
133
+    public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
134
+        logger.info("xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
135
+        return jobHandlerRepository.put(name, jobHandler);
136
+    }
137
+    public static IJobHandler loadJobHandler(String name){
138
+        return jobHandlerRepository.get(name);
139
+    }
140
+    private static void initJobHandlerRepository(ApplicationContext applicationContext){
122
         // init job handler action
141
         // init job handler action
123
         Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHander.class);
142
         Map<String, Object> serviceBeanMap = applicationContext.getBeansWithAnnotation(JobHander.class);
124
 
143
 
134
                 }
153
                 }
135
             }
154
             }
136
         }
155
         }
137
-	}
138
-
139
-    // ---------------------------------- destory job executor ------------------------------------
140
-    @Override
141
-    public void onApplicationEvent(ApplicationEvent applicationEvent) {
142
-        if(applicationEvent instanceof ContextClosedEvent){
143
-            // TODO
144
-        }
145
     }
156
     }
146
 
157
 
147
-    // ---------------------------------- job handler repository
148
-    private static ConcurrentHashMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
149
-    public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
150
-        logger.info("xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
151
-        return jobHandlerRepository.put(name, jobHandler);
152
-    }
153
-    public static IJobHandler loadJobHandler(String name){
154
-        return jobHandlerRepository.get(name);
155
-    }
156
 
158
 
157
-    // ---------------------------------- job thread repository
159
+    // ---------------------------------- job thread repository ------------------------------------
158
     private static ConcurrentHashMap<Integer, JobThread> JobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
160
     private static ConcurrentHashMap<Integer, JobThread> JobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
159
     public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
161
     public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
160
         JobThread newJobThread = new JobThread(jobId, handler);
162
         JobThread newJobThread = new JobThread(jobId, handler);

+ 5 - 5
xxl-job-core/src/main/java/com/xxl/job/core/glue/GlueFactory.java Ver fichero

52
 				try {
52
 				try {
53
 					Resource resource = AnnotationUtils.getAnnotation(field, Resource.class);
53
 					Resource resource = AnnotationUtils.getAnnotation(field, Resource.class);
54
 					if (resource.name()!=null && resource.name().length()>0){
54
 					if (resource.name()!=null && resource.name().length()>0){
55
-						fieldBean = XxlJobExecutor.applicationContext.getBean(resource.name());
55
+						fieldBean = XxlJobExecutor.getApplicationContext().getBean(resource.name());
56
 					} else {
56
 					} else {
57
-						fieldBean = XxlJobExecutor.applicationContext.getBean(field.getName());
57
+						fieldBean = XxlJobExecutor.getApplicationContext().getBean(field.getName());
58
 					}
58
 					}
59
 				} catch (Exception e) {
59
 				} catch (Exception e) {
60
 				}
60
 				}
61
 				if (fieldBean==null ) {
61
 				if (fieldBean==null ) {
62
-					fieldBean = XxlJobExecutor.applicationContext.getBean(field.getType());
62
+					fieldBean = XxlJobExecutor.getApplicationContext().getBean(field.getType());
63
 				}
63
 				}
64
 			} else if (AnnotationUtils.getAnnotation(field, Autowired.class) != null) {
64
 			} else if (AnnotationUtils.getAnnotation(field, Autowired.class) != null) {
65
 				Qualifier qualifier = AnnotationUtils.getAnnotation(field, Qualifier.class);
65
 				Qualifier qualifier = AnnotationUtils.getAnnotation(field, Qualifier.class);
66
 				if (qualifier!=null && qualifier.value()!=null && qualifier.value().length()>0) {
66
 				if (qualifier!=null && qualifier.value()!=null && qualifier.value().length()>0) {
67
-					fieldBean = XxlJobExecutor.applicationContext.getBean(qualifier.value());
67
+					fieldBean = XxlJobExecutor.getApplicationContext().getBean(qualifier.value());
68
 				} else {
68
 				} else {
69
-					fieldBean = XxlJobExecutor.applicationContext.getBean(field.getType());
69
+					fieldBean = XxlJobExecutor.getApplicationContext().getBean(field.getType());
70
 				}
70
 				}
71
 			}
71
 			}
72
 			
72
 			

+ 3 - 4
xxl-job-core/src/main/java/com/xxl/job/core/handler/impl/ScriptJobHandler.java Ver fichero

1
 package com.xxl.job.core.handler.impl;
1
 package com.xxl.job.core.handler.impl;
2
 
2
 
3
 import com.xxl.job.core.biz.model.ReturnT;
3
 import com.xxl.job.core.biz.model.ReturnT;
4
-import com.xxl.job.core.executor.XxlJobExecutor;
5
 import com.xxl.job.core.glue.GlueTypeEnum;
4
 import com.xxl.job.core.glue.GlueTypeEnum;
6
 import com.xxl.job.core.handler.IJobHandler;
5
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.log.XxlJobFileAppender;
6
 import com.xxl.job.core.log.XxlJobFileAppender;
37
         String scriptFileName = null;
36
         String scriptFileName = null;
38
         if (GlueTypeEnum.GLUE_SHELL == glueType) {
37
         if (GlueTypeEnum.GLUE_SHELL == glueType) {
39
             cmd = "bash";
38
             cmd = "bash";
40
-            scriptFileName = XxlJobExecutor.logPath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".sh");
39
+            scriptFileName = XxlJobFileAppender.logPath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".sh");
41
         } else if (GlueTypeEnum.GLUE_PYTHON == glueType) {
40
         } else if (GlueTypeEnum.GLUE_PYTHON == glueType) {
42
             cmd = "python";
41
             cmd = "python";
43
-            scriptFileName = XxlJobExecutor.logPath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".py");
42
+            scriptFileName = XxlJobFileAppender.logPath.concat("gluesource/").concat(String.valueOf(jobId)).concat("_").concat(String.valueOf(glueUpdatetime)).concat(".py");
44
         }
43
         }
45
 
44
 
46
         // make script file
45
         // make script file
47
         ScriptUtil.markScriptFile(scriptFileName, gluesource);
46
         ScriptUtil.markScriptFile(scriptFileName, gluesource);
48
 
47
 
49
         // log file
48
         // log file
50
-        String logFileName = XxlJobExecutor.logPath.concat(XxlJobFileAppender.contextHolder.get());
49
+        String logFileName = XxlJobFileAppender.logPath.concat(XxlJobFileAppender.contextHolder.get());
51
 
50
 
52
         // invoke
51
         // invoke
53
         XxlJobLogger.log("----------- script file:"+ scriptFileName +" -----------");
52
         XxlJobLogger.log("----------- script file:"+ scriptFileName +" -----------");

+ 5 - 5
xxl-job-core/src/main/java/com/xxl/job/core/log/XxlJobFileAppender.java Ver fichero

1
 package com.xxl.job.core.log;
1
 package com.xxl.job.core.log;
2
 
2
 
3
 import com.xxl.job.core.biz.model.LogResult;
3
 import com.xxl.job.core.biz.model.LogResult;
4
-import com.xxl.job.core.executor.XxlJobExecutor;
5
 import org.slf4j.Logger;
4
 import org.slf4j.Logger;
6
 import org.slf4j.LoggerFactory;
5
 import org.slf4j.LoggerFactory;
7
 
6
 
20
 	//public static ThreadLocal<String> contextHolder = new ThreadLocal<String>();
19
 	//public static ThreadLocal<String> contextHolder = new ThreadLocal<String>();
21
 	public static final InheritableThreadLocal<String> contextHolder = new InheritableThreadLocal<String>();
20
 	public static final InheritableThreadLocal<String> contextHolder = new InheritableThreadLocal<String>();
22
 	public static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
21
 	public static final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
23
-	
22
+	public static String logPath = "/data/applogs/xxl-job/jobhandler/";
23
+
24
 	/**
24
 	/**
25
 	 * log filename: yyyy-MM-dd/9999.log
25
 	 * log filename: yyyy-MM-dd/9999.log
26
 	 *
26
 	 *
31
 	public static String makeLogFileName(Date triggerDate, int logId) {
31
 	public static String makeLogFileName(Date triggerDate, int logId) {
32
 
32
 
33
         // filePath/
33
         // filePath/
34
-        File filePathDir = new File(XxlJobExecutor.logPath);
34
+        File filePathDir = new File(logPath);
35
         if (!filePathDir.exists()) {
35
         if (!filePathDir.exists()) {
36
             filePathDir.mkdirs();
36
             filePathDir.mkdirs();
37
         }
37
         }
66
 		if (logFileName==null || logFileName.trim().length()==0) {
66
 		if (logFileName==null || logFileName.trim().length()==0) {
67
 			return;
67
 			return;
68
 		}
68
 		}
69
-		File logFile = new File(XxlJobExecutor.logPath, logFileName);
69
+		File logFile = new File(logPath, logFileName);
70
 
70
 
71
 		if (!logFile.exists()) {
71
 		if (!logFile.exists()) {
72
 			try {
72
 			try {
111
 		if (logFileName==null || logFileName.trim().length()==0) {
111
 		if (logFileName==null || logFileName.trim().length()==0) {
112
             return new LogResult(fromLineNum, 0, "readLog fail, logFile not found", true);
112
             return new LogResult(fromLineNum, 0, "readLog fail, logFile not found", true);
113
 		}
113
 		}
114
-		File logFile = new File(XxlJobExecutor.logPath, logFileName);
114
+		File logFile = new File(logPath, logFileName);
115
 
115
 
116
 		if (!logFile.exists()) {
116
 		if (!logFile.exists()) {
117
             return new LogResult(fromLineNum, 0, "readLog fail, logFile not exists", true);
117
             return new LogResult(fromLineNum, 0, "readLog fail, logFile not exists", true);

+ 17 - 1
xxl-job-core/src/main/java/com/xxl/job/core/rpc/netcom/jetty/server/JettyServer.java Ver fichero

1
 package com.xxl.job.core.rpc.netcom.jetty.server;
1
 package com.xxl.job.core.rpc.netcom.jetty.server;
2
 
2
 
3
 import com.xxl.job.core.thread.ExecutorRegistryThread;
3
 import com.xxl.job.core.thread.ExecutorRegistryThread;
4
+import com.xxl.job.core.thread.TriggerCallbackThread;
4
 import org.eclipse.jetty.server.Connector;
5
 import org.eclipse.jetty.server.Connector;
5
 import org.eclipse.jetty.server.Handler;
6
 import org.eclipse.jetty.server.Handler;
6
 import org.eclipse.jetty.server.Server;
7
 import org.eclipse.jetty.server.Server;
38
 				server.setHandler(handlerc);
39
 				server.setHandler(handlerc);
39
 
40
 
40
 				try {
41
 				try {
41
-					// Start the server
42
+					// Start server
42
 					server.start();
43
 					server.start();
43
 					logger.info(">>>>>>>>>>>> xxl-job jetty server start success at port:{}.", port);
44
 					logger.info(">>>>>>>>>>>> xxl-job jetty server start success at port:{}.", port);
45
+
46
+					// Start Registry-Server
44
 					ExecutorRegistryThread.getInstance().start(port, ip, appName);
47
 					ExecutorRegistryThread.getInstance().start(port, ip, appName);
48
+
49
+					// Start Callback-Server
50
+					TriggerCallbackThread.getInstance().start();
51
+
45
 					server.join();	// block until thread stopped
52
 					server.join();	// block until thread stopped
46
 					logger.info(">>>>>>>>>>> xxl-rpc server join success, netcon={}, port={}", JettyServer.class.getName(), port);
53
 					logger.info(">>>>>>>>>>> xxl-rpc server join success, netcon={}, port={}", JettyServer.class.getName(), port);
47
 				} catch (Exception e) {
54
 				} catch (Exception e) {
56
 	}
63
 	}
57
 
64
 
58
 	public void destroy() {
65
 	public void destroy() {
66
+
67
+		// destroy server
59
 		if (server != null) {
68
 		if (server != null) {
60
 			try {
69
 			try {
61
 				server.stop();
70
 				server.stop();
67
 		if (thread.isAlive()) {
76
 		if (thread.isAlive()) {
68
 			thread.interrupt();
77
 			thread.interrupt();
69
 		}
78
 		}
79
+
80
+		// destroy Registry-Server
81
+		ExecutorRegistryThread.getInstance().toStop();
82
+
83
+		// destroy Callback-Server
84
+		TriggerCallbackThread.getInstance().toStop();
85
+
70
 		logger.info(">>>>>>>>>>> xxl-rpc server destroy success, netcon={}", JettyServer.class.getName());
86
 		logger.info(">>>>>>>>>>> xxl-rpc server destroy success, netcon={}", JettyServer.class.getName());
71
 	}
87
 	}
72
 
88
 

+ 2 - 2
xxl-job-core/src/main/java/com/xxl/job/core/util/ScriptUtil.java Ver fichero

1
 package com.xxl.job.core.util;
1
 package com.xxl.job.core.util;
2
 
2
 
3
-import com.xxl.job.core.executor.XxlJobExecutor;
3
+import com.xxl.job.core.log.XxlJobFileAppender;
4
 import org.apache.commons.exec.CommandLine;
4
 import org.apache.commons.exec.CommandLine;
5
 import org.apache.commons.exec.DefaultExecutor;
5
 import org.apache.commons.exec.DefaultExecutor;
6
 import org.apache.commons.exec.PumpStreamHandler;
6
 import org.apache.commons.exec.PumpStreamHandler;
28
      */
28
      */
29
     public static void markScriptFile(String scriptFileName, String content) throws IOException {
29
     public static void markScriptFile(String scriptFileName, String content) throws IOException {
30
         // filePath/
30
         // filePath/
31
-        File filePathDir = new File(XxlJobExecutor.logPath);
31
+        File filePathDir = new File(XxlJobFileAppender.logPath);
32
         if (!filePathDir.exists()) {
32
         if (!filePathDir.exists()) {
33
             filePathDir.mkdirs();
33
             filePathDir.mkdirs();
34
         }
34
         }