修改默认线程数

Author: 胡贵

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;
             }
         }
     }

pom.xml

@@ -39,6 +39,18 @@
         <druid.version>1.0.14</druid.version>
         <mysql.version>5.1.26</mysql.version>
     </properties>
+    <distributionManagement>
+        <repository>
+            <id>nexus-releases</id>
+            <name>Local Nexus Repository</name>
+            <url>http://maven.superboss.cc:8081/nexus/content/repositories/releases</url>
+        </repository>
+        <snapshotRepository>
+            <id>nexus-snapshots</id>
+            <name>Local Nexus Repository</name>
+            <url>http://maven.superboss.cc:8081/nexus/content/repositories/snapshots</url>
+        </snapshotRepository>
+    </distributionManagement>
 
     <dependencyManagement>
         <dependencies>