Merge pull request #195 from qq254963746/develop

增加有中断接口的   InterruptibleJobRunner

Author: qq254963746

lts-tasktracker/src/main/java/com/lts/tasktracker/runner/JobRunnerDelegate.java

@@ -35,38 +35,33 @@ public class JobRunnerDelegate implements Runnable {
     private TaskTrackerMonitor monitor;
     private Interruptible interruptor;
     private JobRunner curJobRunner;
-    private boolean isInterruptibleJobRunner = false;
+    private boolean interrupted = false;
 
     public JobRunnerDelegate(TaskTrackerAppContext appContext,
                              JobWrapper jobWrapper, RunnerCallback callback) {
         this.appContext = appContext;
         this.callback = callback;
         this.jobWrapper = jobWrapper;
 
-        this.isInterruptibleJobRunner = isInterruptibleJobRunner(this.appContext);
         this.logger = (BizLoggerAdapter) BizLoggerFactory.getLogger(
                 appContext.getBizLogLevel(),
                 appContext.getRemotingClient(), appContext);
         monitor = (TaskTrackerMonitor) appContext.getMonitor();
 
-        if (isInterruptibleJobRunner()) {
-            this.interruptor = new Interruptible() {
-                @Override
-                public void interrupt() {
-                    JobRunnerDelegate.this.interrupt();
-                }
-            };
-        }
+        this.interruptor = new Interruptible() {
+            @Override
+            public void interrupt() {
+                JobRunnerDelegate.this.interrupt();
+            }
+        };
     }
 
     @Override
     public void run() {
         try {
-            if (isInterruptibleJobRunner()) {
-                blockedOn(interruptor);
-                if (Thread.currentThread().isInterrupted()) {
-                    interruptor.interrupt();
-                }
+            blockedOn(interruptor);
+            if (Thread.currentThread().isInterrupted()) {
+                interruptor.interrupt();
             }
 
             LtsLoggerFactory.setLogger(logger);
@@ -116,32 +111,30 @@ public void run() {
                 } catch (Throwable t) {
                     LOGGER.warn("monitor error:" + t.getMessage(), t);
                 }
-
+                if (isInterrupted()) {
+                    // 如果当前线程被阻断了,那么也就不接受新任务了
+                    response.setReceiveNewJob(false);
+                }
                 this.jobWrapper = callback.runComplete(response);
 
             }
         } finally {
             LtsLoggerFactory.remove();
 
-            if (isInterruptibleJobRunner()) {
-                blockedOn(null);
-            }
+            blockedOn(null);
         }
     }
 
     private void interrupt() {
+        interrupted = true;
+
         if (this.curJobRunner != null && this.curJobRunner instanceof InterruptibleJobRunner) {
             ((InterruptibleJobRunner) this.curJobRunner).interrupt();
         }
     }
 
-    private static boolean isInterruptibleJobRunner(TaskTrackerAppContext appContext) {
-        Class<?> jobRunnerClass = appContext.getJobRunnerClass();
-        return jobRunnerClass != null && InterruptibleJobRunner.class.isAssignableFrom(appContext.getJobRunnerClass());
-    }
-
-    private boolean isInterruptibleJobRunner() {
-        return this.isInterruptibleJobRunner;
+    private boolean isInterrupted() {
+        return this.interrupted;
     }
 
     private void monitor(Action action) {