优化 客户端 通知 负债均衡

Author: qq254963746

job-core/src/main/java/com/lts/job/core/remoting/HeartBeatMonitor.java

@@ -1,6 +1,8 @@
 package com.lts.job.core.remoting;
 
+import com.lts.job.core.cluster.NodeType;
 import com.lts.job.core.exception.JobTrackerNotFoundException;
+import com.lts.job.core.support.Application;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,8 +28,13 @@ public HeartBeatMonitor(RemotingClientDelegate remotingClient) {
     }
 
     public void start() {
+        // 设置 JobClient 心跳10s一次，TaskTracker 5s一次
+        int delay = 5;
+        if(NodeType.CLIENT.equals(Application.Config.getNodeType())){
+            delay = 10;
+        }
         HEART_BEAT_EXECUTOR_SERVICE.scheduleWithFixedDelay(
-                new HeartBeatRunner(), 5, 5, TimeUnit.SECONDS);  // 5s发送一次心跳
+                new HeartBeatRunner(), 5, delay, TimeUnit.SECONDS);
     }
 
     public void destroy() {

job-core/src/main/java/com/lts/job/core/repository/JobFeedbackQueueMongoRepository.java

@@ -16,6 +16,11 @@ public void delJobFeedback(String id) {
         ds.delete(query);
     }
 
+    public long count(){
+        Query<JobFeedbackQueuePo> query = createQuery();
+        return ds.getCount(query);
+    }
+
     public List<JobFeedbackQueuePo> get(int offset, int limit){
         Query<JobFeedbackQueuePo> query = createQuery();
         query.order("gmtCreated").offset(offset).limit(limit);

job-core/target/maven-archiver/pom.properties

@@ -83,6 +83,13 @@ private class Runner implements Runnable {
         @Override
         public void run() {
             try {
+
+                long count = jobFeedbackQueueMongoRepository.count();
+                if (count == 0) {
+                    return;
+                }
+                LOGGER.info("一共有{}个完成的任务要通知客户端.", count);
+
                 List<JobFeedbackQueuePo> jobFeedbackQueuePos;
                 int offset = 0;
                 int limit = 5;
@@ -91,7 +98,6 @@ public void run() {
                     if (CollectionUtils.isEmpty(jobFeedbackQueuePos)) {
                         return;
                     }
-                    LOGGER.info("一共有{}个完成的任务要通知客户端.", jobFeedbackQueuePos.size());
                     List<JobResult> jobResults = new ArrayList<JobResult>(jobFeedbackQueuePos.size());
                     for (JobFeedbackQueuePo jobFeedbackQueuePo : jobFeedbackQueuePos) {
                         jobResults.add(jobFeedbackQueuePo);

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

@@ -9,11 +9,14 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.ArrayList;
+import java.util.List;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadLocalRandom;
 
 /**
  * @author Robert HG (254963746@qq.com) on 8/17/14.
- * 客户端节点管理
+ *         客户端节点管理
  */
 public class JobClientManager {
 
@@ -36,7 +39,7 @@ public void addNode(Node node) {
         ChannelWrapper channel = ChannelManager.getChannel(node.getGroup(), node.getNodeType(), node.getIdentity());
         ConcurrentHashSet<JobClientNode> jobClientNodes = NODE_MAP.get(node.getGroup());
 
-        synchronized (NODE_MAP){
+        synchronized (NODE_MAP) {
             if (jobClientNodes == null) {
                 jobClientNodes = new ConcurrentHashSet<JobClientNode>();
                 NODE_MAP.put(node.getGroup(), jobClientNodes);
@@ -58,7 +61,7 @@ public void removeNode(Node node) {
         ConcurrentHashSet<JobClientNode> jobClientNodes = NODE_MAP.get(node.getGroup());
         if (jobClientNodes != null && jobClientNodes.size() != 0) {
             for (JobClientNode jobClientNode : jobClientNodes) {
-                if(node.getIdentity().equals(jobClientNode.getIdentity())){
+                if (node.getIdentity().equals(jobClientNode.getIdentity())) {
                     LOGGER.info("删除JobClient节点:{}", jobClientNode);
                     jobClientNodes.remove(jobClientNode);
                 }
@@ -68,31 +71,49 @@ public void removeNode(Node node) {
 
     /**
      * 得到 可用的 客户端节点
+     *
      * @param nodeGroup
      * @return
      */
     public JobClientNode getAvailableJobClient(String nodeGroup) {
 
         ConcurrentHashSet<JobClientNode> jobClientNodes = NODE_MAP.get(nodeGroup);
+
         if (jobClientNodes == null || jobClientNodes.size() == 0) {
             return null;
         }
 
-        for (JobClientNode jobClientNode : jobClientNodes) {
+        int size = jobClientNodes.size();
+        int index = getRandomIndex(size);
+
+        List<JobClientNode> list = new ArrayList<JobClientNode>(jobClientNodes);
+
+        JobClientNode jobClientNode = null;
+        int retry = 0;
+        while (jobClientNode == null && retry < size) {
+            jobClientNode = list.get(index);
             // 如果 channel 已经关闭, 更新channel, 如果没有channel, 略过
-            if (jobClientNode.getChannel() == null || jobClientNode.getChannel().isClosed()) {
+            if (jobClientNode != null && (jobClientNode.getChannel() == null || jobClientNode.getChannel().isClosed())) {
                 ChannelWrapper channel = ChannelManager.getChannel(jobClientNode.getNodeGroup(), NodeType.CLIENT, jobClientNode.getIdentity());
                 if (channel != null) {
                     // 更新channel
                     jobClientNode.setChannel(channel);
                 } else {
-                    continue;
+                    jobClientNode = null;
                 }
             }
-
-            return jobClientNode;
+            index = (index + 1) % size;
+            retry++;
         }
 
-        return null;
+        return jobClientNode;
+    }
+
+    private int getRandomIndex(int size) {
+        int min = 1;
+        int max = size;
+        ThreadLocalRandom random = ThreadLocalRandom.current();
+        return random.nextInt(max) % (max - min + 1) + min - 1;
     }
+
 }