Bläddra i källkod

阻塞处理策略

xueli.xue 8 år sedan
förälder
incheckning
0888b2d5a4

+ 1 - 0
xxl-job-admin/src/main/java/com/xxl/job/admin/core/jobbean/RemoteHttpJobBean.java Visa fil

61
 		triggerParam.setJobId(jobInfo.getId());
61
 		triggerParam.setJobId(jobInfo.getId());
62
 		triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
62
 		triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
63
 		triggerParam.setExecutorParams(jobInfo.getExecutorParam());
63
 		triggerParam.setExecutorParams(jobInfo.getExecutorParam());
64
+        triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy());
64
 		triggerParam.setGlueType(jobInfo.getGlueType());
65
 		triggerParam.setGlueType(jobInfo.getGlueType());
65
 		triggerParam.setGlueSource(jobInfo.getGlueSource());
66
 		triggerParam.setGlueSource(jobInfo.getGlueSource());
66
 		triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());
67
 		triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime());

+ 4 - 2
xxl-job-core/src/main/java/com/xxl/job/core/biz/impl/ExecutorBizImpl.java Visa fil

4
 import com.xxl.job.core.biz.model.LogResult;
4
 import com.xxl.job.core.biz.model.LogResult;
5
 import com.xxl.job.core.biz.model.ReturnT;
5
 import com.xxl.job.core.biz.model.ReturnT;
6
 import com.xxl.job.core.biz.model.TriggerParam;
6
 import com.xxl.job.core.biz.model.TriggerParam;
7
+import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
7
 import com.xxl.job.core.executor.XxlJobExecutor;
8
 import com.xxl.job.core.executor.XxlJobExecutor;
8
 import com.xxl.job.core.glue.GlueFactory;
9
 import com.xxl.job.core.glue.GlueFactory;
9
 import com.xxl.job.core.glue.GlueTypeEnum;
10
 import com.xxl.job.core.glue.GlueTypeEnum;
128
         }
129
         }
129
 
130
 
130
         // push data to queue
131
         // push data to queue
131
-        jobThread.pushTriggerQueue(triggerParam);
132
-        return ReturnT.SUCCESS;
132
+        ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
133
+        ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam, blockStrategy);
134
+        return pushResult;
133
     }
135
     }
134
 
136
 
135
 }
137
 }

+ 9 - 0
xxl-job-core/src/main/java/com/xxl/job/core/biz/model/TriggerParam.java Visa fil

13
 
13
 
14
     private String executorHandler;
14
     private String executorHandler;
15
     private String executorParams;
15
     private String executorParams;
16
+    private String executorBlockStrategy;
16
 
17
 
17
     private String glueType;
18
     private String glueType;
18
     private String glueSource;
19
     private String glueSource;
47
         this.executorParams = executorParams;
48
         this.executorParams = executorParams;
48
     }
49
     }
49
 
50
 
51
+    public String getExecutorBlockStrategy() {
52
+        return executorBlockStrategy;
53
+    }
54
+
55
+    public void setExecutorBlockStrategy(String executorBlockStrategy) {
56
+        this.executorBlockStrategy = executorBlockStrategy;
57
+    }
58
+
50
     public String getGlueType() {
59
     public String getGlueType() {
51
         return glueType;
60
         return glueType;
52
     }
61
     }

+ 29 - 4
xxl-job-core/src/main/java/com/xxl/job/core/thread/JobThread.java Visa fil

3
 import com.xxl.job.core.biz.model.HandleCallbackParam;
3
 import com.xxl.job.core.biz.model.HandleCallbackParam;
4
 import com.xxl.job.core.biz.model.ReturnT;
4
 import com.xxl.job.core.biz.model.ReturnT;
5
 import com.xxl.job.core.biz.model.TriggerParam;
5
 import com.xxl.job.core.biz.model.TriggerParam;
6
+import com.xxl.job.core.enums.ExecutorBlockStrategyEnum;
6
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.handler.IJobHandler;
7
 import com.xxl.job.core.log.XxlJobFileAppender;
8
 import com.xxl.job.core.log.XxlJobFileAppender;
8
 import com.xxl.job.core.log.XxlJobLogger;
9
 import com.xxl.job.core.log.XxlJobLogger;
31
 	private boolean toStop = false;
32
 	private boolean toStop = false;
32
 	private String stopReason;
33
 	private String stopReason;
33
 
34
 
35
+    private boolean running = false;
36
+
37
+
34
 	public JobThread(IJobHandler handler) {
38
 	public JobThread(IJobHandler handler) {
35
 		this.handler = handler;
39
 		this.handler = handler;
36
 		triggerQueue = new LinkedBlockingQueue<TriggerParam>();
40
 		triggerQueue = new LinkedBlockingQueue<TriggerParam>();
40
 		return handler;
44
 		return handler;
41
 	}
45
 	}
42
 
46
 
43
-	public void pushTriggerQueue(TriggerParam triggerParam) {
47
+	public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam, ExecutorBlockStrategyEnum blockStrategy) {
48
+		// avoid repeat
44
 		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
49
 		if (triggerLogIdSet.contains(triggerParam.getLogId())) {
45
 			logger.debug("repeate trigger job, logId:{}", triggerParam.getLogId());
50
 			logger.debug("repeate trigger job, logId:{}", triggerParam.getLogId());
46
-			return;
51
+			return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
52
+		}
53
+
54
+		// block strategy
55
+		if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
56
+            // discard when running
57
+            if (running) {
58
+                return new ReturnT<String>(ReturnT.FAIL_CODE, "任务阻塞:"+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
59
+            }
60
+		} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
61
+            // kill running old and clear queue
62
+            if (running) {
63
+                this.interrupt();
64
+            }
65
+            triggerQueue.clear();
66
+            triggerLogIdSet.clear();
67
+		} else {
68
+            // just add to queue
47
 		}
69
 		}
48
 
70
 
49
 		triggerLogIdSet.add(triggerParam.getLogId());
71
 		triggerLogIdSet.add(triggerParam.getLogId());
50
 		triggerQueue.add(triggerParam);
72
 		triggerQueue.add(triggerParam);
73
+        return ReturnT.SUCCESS;
51
 	}
74
 	}
52
 
75
 
53
 	public void toStop(String stopReason) {
76
 	public void toStop(String stopReason) {
59
 		this.toStop = true;
82
 		this.toStop = true;
60
 		this.stopReason = stopReason;
83
 		this.stopReason = stopReason;
61
 	}
84
 	}
62
-	
63
-	int i = 1;
85
+
86
+
64
 	@Override
87
 	@Override
65
 	public void run() {
88
 	public void run() {
66
 		while(!toStop){
89
 		while(!toStop){
90
+			running = false;
67
 			try {
91
 			try {
68
 				// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
92
 				// to check toStop signal, we need cycle, so wo cannot use queue.take(), instand of poll(timeout)
69
 				TriggerParam triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
93
 				TriggerParam triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);
70
 				if (triggerParam!=null) {
94
 				if (triggerParam!=null) {
95
+					running = true;
71
 					triggerLogIdSet.remove(triggerParam.getLogId());
96
 					triggerLogIdSet.remove(triggerParam.getLogId());
72
 					
97
 					
73
 					// parse param
98
 					// parse param