diff --git a/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdAcceptor.java b/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdAcceptor.java index ad2a7181d..f94bc55a0 100644 --- a/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdAcceptor.java +++ b/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdAcceptor.java @@ -11,6 +11,7 @@ import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; +import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.atomic.AtomicBoolean; /** @@ -32,7 +33,15 @@ public HttpCmdAcceptor(ServerSocket serverSocket, HttpCmdContext context) { this.executorService = new ThreadPoolExecutor(Constants.AVAILABLE_PROCESSOR, Constants.AVAILABLE_PROCESSOR, 0L, TimeUnit.MILLISECONDS, - new LinkedBlockingQueue(100), new ThreadPoolExecutor.DiscardPolicy()); + new LinkedBlockingQueue(100), new RejectedExecutionHandler() { + @Override + public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) { + if (r instanceof HttpCmdExecutor) { + HttpCmdExecutor httpCmdExecutor = (HttpCmdExecutor)r; + httpCmdExecutor.abort(); + } + } + }); } public void start() { diff --git a/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdExecutor.java b/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdExecutor.java index 8588e1c4f..221384efd 100644 --- a/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdExecutor.java +++ b/lts-core/src/main/java/com/github/ltsopensource/cmd/HttpCmdExecutor.java @@ -51,6 +51,11 @@ public void run() { } catch (Throwable t) { LOGGER.error("Error When Execute Command", t); sendError(HTTP_INTERNALERROR, JSON.toJSONString(HttpCmdResponse.newResponse(false, "Error:" + t.getMessage())), false); + } finally { + try { + socket.close(); + } catch (IOException e) { + } } } @@ -111,6 +116,12 @@ private HttpCmdRequest parseRequest() throws Exception { return resolveRequest(uri, params); } + public void abort() { + try { + this.socket.close(); + } catch (IOException e) { + } + } protected static HttpCmdRequest resolveRequest(String uri, Properties params) { diff --git a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutableDeadJobChecker.java b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutableDeadJobChecker.java index 6bff7244f..47fa43eef 100644 --- a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutableDeadJobChecker.java +++ b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutableDeadJobChecker.java @@ -29,7 +29,7 @@ public class ExecutableDeadJobChecker { // 1 分钟还锁着的,说明是有问题的 private static final long MAX_TIME_OUT = 60 * 1000; - private final ScheduledExecutorService FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutableJobQueue-Fix-Executor", true)); + private ScheduledExecutorService FIXED_EXECUTOR_SERVICE; private JobTrackerAppContext appContext; @@ -43,6 +43,7 @@ public ExecutableDeadJobChecker(JobTrackerAppContext appContext) { public void start() { try { if (start.compareAndSet(false, true)) { + FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutableJobQueue-Fix-Executor", true)); scheduledFuture = FIXED_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runnable() { @Override public void run() { diff --git a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutingDeadJobChecker.java b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutingDeadJobChecker.java index 5878295f3..31a9c3ed9 100644 --- a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutingDeadJobChecker.java +++ b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/ExecutingDeadJobChecker.java @@ -50,7 +50,7 @@ public class ExecutingDeadJobChecker { private static final Logger LOGGER = LoggerFactory.getLogger(ExecutingDeadJobChecker.class); - private final ScheduledExecutorService FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutingJobQueue-Fix-Executor", true)); + private ScheduledExecutorService FIXED_EXECUTOR_SERVICE; private JobTrackerAppContext appContext; private JobTrackerMStatReporter stat; @@ -73,6 +73,7 @@ public void start() { fixCheckPeriodSeconds = 5 * 60; } + FIXED_EXECUTOR_SERVICE = Executors.newScheduledThreadPool(1, new NamedThreadFactory("LTS-ExecutingJobQueue-Fix-Executor", true)); scheduledFuture = FIXED_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runnable() { @Override public void run() { diff --git a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/FeedbackJobSendChecker.java b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/FeedbackJobSendChecker.java index d5425b0b6..a869ce5ac 100644 --- a/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/FeedbackJobSendChecker.java +++ b/lts-jobtracker/src/main/java/com/github/ltsopensource/jobtracker/support/checker/FeedbackJobSendChecker.java @@ -29,7 +29,7 @@ public class FeedbackJobSendChecker { private static final Logger LOGGER = LoggerFactory.getLogger(FeedbackJobSendChecker.class); - private ScheduledExecutorService RETRY_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LTS-FeedbackJobSend-Executor", true)); + private ScheduledExecutorService RETRY_EXECUTOR_SERVICE; private ScheduledFuture scheduledFuture; private AtomicBoolean start = new AtomicBoolean(false); private ClientNotifier clientNotifier; @@ -68,6 +68,7 @@ public void handleFailed(List jobResults) { public void start() { try { if (start.compareAndSet(false, true)) { + RETRY_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("LTS-FeedbackJobSend-Executor", true)); scheduledFuture = RETRY_EXECUTOR_SERVICE.scheduleWithFixedDelay(new Runner() , 30, 30, TimeUnit.SECONDS); }