剩余 mysql 的dead lock 没有解决

Author: 胡贵

job-admin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>job-parent</artifactId>
         <groupId>com.lts</groupId>
-        <version>1.4.0-SNAPSHOT</version>
+        <version>1.4.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>war</packaging>

job-client/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>job-parent</artifactId>
         <groupId>com.lts</groupId>
-        <version>1.4.0-SNAPSHOT</version>
+        <version>1.4.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 

job-core/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>job-parent</artifactId>
         <groupId>com.lts</groupId>
-        <version>1.4.0-SNAPSHOT</version>
+        <version>1.4.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>

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

@@ -51,7 +51,10 @@ private void _addNode(Node node) {
         List<Node> nodeList = NODES.get(node.getNodeType());
         if (CollectionUtils.isEmpty(nodeList)) {
             nodeList = new CopyOnWriteArrayList<Node>();
-            NODES.put(node.getNodeType(), nodeList);
+            List<Node> oldNodeList = NODES.putIfAbsent(node.getNodeType(), nodeList);
+            if (oldNodeList != null) {
+                nodeList = oldNodeList;
+            }
         }
         nodeList.add(node);
         LOGGER.info("add node {}", node);

job-core/src/main/java/com/lts/job/core/loadbalance/RoundbinLoadBalance.java

@@ -0,0 +1,16 @@
+package com.lts.job.core.loadbalance;
+
+import java.util.List;
+
+/**
+ * Created by hugui on 5/31/15.
+ */
+public class RoundbinLoadBalance extends AbstractLoadBalance{
+
+
+
+    @Override
+    protected <S> S doSelect(List<S> shards, String seed) {
+        return null;
+    }
+}

job-core/src/main/java/com/lts/job/core/util/JobQueueUtils.java

@@ -19,7 +19,7 @@ public static String getExecutableQueueName(String taskTrackerNodeGroup) {
      * 在数据库中就是表名, jobClientNodeGroup 是 JobClient 的 nodeGroup
      */
     public static String getFeedbackQueueName(String jobClientNodeGroup) {
-        return "lt_feedback_job_queue_".concat(jobClientNodeGroup);
+        return "lts_feedback_job_queue_".concat(jobClientNodeGroup);
     }
 
     public static final String CRON_JOB_QUEUE = "lts_cron_job_queue";

job-core/src/main/java/com/lts/job/ec/injvm/InjvmEventCenter.java

@@ -9,6 +9,7 @@
 import com.lts.job.ec.EventInfo;
 import com.lts.job.ec.EventSubscriber;
 
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
@@ -32,9 +33,10 @@ public class InjvmEventCenter implements EventCenter {
     public void subscribe(String topic, EventSubscriber subscriber) {
         Set<EventSubscriber> subscribers = ecMap.get(topic);
         if (subscribers == null) {
-            synchronized (ecMap) {
-                subscribers = new ConcurrentHashSet<EventSubscriber>();
-                ecMap.put(topic, subscribers);
+            subscribers = new ConcurrentHashSet<EventSubscriber>();
+            Set<EventSubscriber> oldSubscribers = ecMap.putIfAbsent(topic, subscribers);
+            if (oldSubscribers != null) {
+                subscribers = oldSubscribers;
             }
         }
         subscribers.add(subscriber);

job-core/src/main/java/com/lts/job/remoting/netty/NettyClientConfig.java

@@ -8,7 +8,7 @@
 public class NettyClientConfig {
     // 处理Server Response/Request
     private int clientWorkerThreads = 4;
-    private int clientCallbackExecutorThreads = Constants.AVAILABLE_PROCESSOR;
+    private int clientCallbackExecutorThreads = Constants.AVAILABLE_PROCESSOR * 2;
     private int clientSelectorThreads = 1;
     private int clientOnewaySemaphoreValue = 256;
     private int clientAsyncSemaphoreValue = 128;

job-core/src/main/java/com/lts/job/remoting/netty/NettyServerConfig.java

@@ -8,7 +8,7 @@
 public class NettyServerConfig {
     private int listenPort = 8888;
     private int serverWorkerThreads = 32;
-    private int serverCallbackExecutorThreads = 0;
+    private int serverCallbackExecutorThreads = Constants.AVAILABLE_PROCESSOR * 2;
     private int serverSelectorThreads = Constants.AVAILABLE_PROCESSOR * 2;
     private int serverOnewaySemaphoreValue = 32;
     private int serverAsyncSemaphoreValue = 64;

job-example/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>job-parent</artifactId>
         <groupId>com.lts</groupId>
-        <version>1.4.0-SNAPSHOT</version>
+        <version>1.4.2-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>job-example</artifactId>