Merge pull request #11 from qq254963746/develop

Develop

Author: qq254963746

job-core/src/main/java/com/lts/job/core/cluster/AbstractClientNode.java

@@ -33,7 +33,7 @@ protected void innerStart() {
         if (defaultProcessor != null) {
 
             remotingClient.registerDefaultProcessor(defaultProcessor,
-                    Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 2,
+                    Executors.newFixedThreadPool(32 + Constants.AVAILABLE_PROCESSOR * 5,
                             new NamedThreadFactory(AbstractClientNode.class.getSimpleName())));
         }
     }

job-core/src/main/java/com/lts/job/core/cluster/AbstractServerNode.java

@@ -30,9 +30,8 @@ protected void innerStart() {
 
         NettyRequestProcessor defaultProcessor = getDefaultProcessor();
         if (defaultProcessor != null) {
-
             remotingServer.registerDefaultProcessor(defaultProcessor,
-                    Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 2, new NamedThreadFactory(AbstractServerNode.class.getSimpleName())));
+                    Executors.newFixedThreadPool(32 + Constants.AVAILABLE_PROCESSOR * 5, new NamedThreadFactory(AbstractServerNode.class.getSimpleName())));
         }
     }
 

job-tracker/src/main/java/com/lts/job/tracker/processor/JobPullProcessor.java

@@ -30,7 +30,7 @@ public class JobPullProcessor extends AbstractProcessor {
     public JobPullProcessor(RemotingServerDelegate remotingServer, JobTrackerApplication application) {
         super(remotingServer, application);
 
-        executor = Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 2
+        executor = Executors.newFixedThreadPool(Constants.AVAILABLE_PROCESSOR * 5
                 , new NamedThreadFactory(JobPullProcessor.class.getSimpleName()));
         jobDistributor = new JobDistributor(application);
     }

job-tracker/src/main/java/com/lts/job/tracker/support/checker/FeedbackJobSendChecker.java

@@ -84,10 +84,16 @@ public void stop() {
         }
     }
 
+    private volatile boolean isRunning = false;
+
     private class Runner implements Runnable {
         @Override
         public void run() {
             try {
+                if (isRunning) {
+                    return;
+                }
+                isRunning = true;
                 long count = application.getJobFeedbackQueue().count();
                 if (count == 0) {
                     return;
@@ -119,6 +125,8 @@ public void run() {
 
             } catch (Throwable t) {
                 LOGGER.error(t.getMessage(), t);
+            } finally {
+                isRunning = false;
             }
         }
     }