jxing 6 tahun lalu
induk
melakukan
fbd62305e4

+ 3 - 1
scheduler/build.gradle

@@ -32,7 +32,9 @@ protobuf {
 //        }
 //    }
 }
-
+tasks.withType(JavaCompile) {
+    options.encoding = "UTF-8"
+}
 dependencies {
     ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
     // kotlin 依赖

+ 2 - 1
scheduler/src/main/java/cn/sagacloud/pojo/ChannelHandlerContextWrapper.java

@@ -15,7 +15,7 @@ import java.util.Map;
 public class ChannelHandlerContextWrapper {
     private ChannelHandlerContext ctx;  // 通信handler
     private long lastRejectTime;        // unix时间片, 精确到秒
-    private final long sendingTimeOut = 10;            // 发送给客户端任务, 如果超过该时间未响应, 则超时, 并将发送的任务状态置为初始状态
+    private final long sendingTimeOut = 15;            // 发送给客户端任务, 如果超过该时间未响应, 则超时, 并将发送的任务状态置为初始状态
     private final long lastRefuseTimeOut = 3600;       // 如果客户端拒绝任务, 则等待lastRefuseTimeOut秒, 才会再分配任务给该客户端
     private long lastRefuseTime = 0L;
     private String clientInfo;             // mac地址
@@ -32,6 +32,7 @@ public class ChannelHandlerContextWrapper {
         if(ctx.channel().isActive()){
             MessageProto.Message msg = MessageUtil.buildMessage(Command.SendTask.name(), toBeSent.getId(), JSON.toJSONString(toBeSent));
             ctx.writeAndFlush(msg);
+            toBeSent.setTask_status(1);
             taskSendingStatusMap.put(toBeSent.getId(), CommonUtil.getTime());
         }
     }

+ 30 - 20
scheduler/src/main/java/cn/sagacloud/server/DispatchTask.java

@@ -72,28 +72,34 @@ public class DispatchTask implements Runnable {
     @Override
     public void run() {
         // 考虑加锁
-        for(long cnt = Long.MAX_VALUE; cnt > Long.MIN_VALUE; --cnt) {
-            // 获取到一个在waiting状态的任务
-            TaskModel toBeSent = getOneAvailableTask();
-            if(toBeSent != null) {
-                // 获取到一个潜在可以接受任务的客户端
-                ChannelHandlerContextWrapper client = getOneAvailableClient();
-                if(client != null) {
-                    // 将获取到的任务发送给选中客户端
-                    client.sendTask(toBeSent);
-                    // 检测sending, sent状态的任务, 是否有超时需要重新下载的任务
-                    checkSendingTimeOut();
-                    checkExecuteTimeOut();
+        try {
+            for (long cnt = Long.MAX_VALUE; cnt > Long.MIN_VALUE; --cnt) {
+                // 获取到一个在waiting状态的任务
+                TaskModel toBeSent = getOneAvailableTask();
+                if (toBeSent != null) {
+                    // 获取到一个潜在可以接受任务的客户端
+                    ChannelHandlerContextWrapper client = getOneAvailableClient();
+                    if (client != null) {
+                        // 将获取到的任务发送给选中客户端
+                        client.sendTask(toBeSent);
+                    }
+                }
+                // 检测sending, sent状态的任务, 是否有超时需要重新下载的任务
+                checkSendingTimeOut();
+                checkExecuteTimeOut();
+                try {
+                    Thread.sleep(1000);
+                } catch (InterruptedException e) {
+                    e.printStackTrace();
+                }
+                if (cnt % 500 == 0L) {
+                    syncTask();
+                    System.out.println(cnt);
                 }
             }
-            try {
-                Thread.sleep(5000);
-            } catch (InterruptedException e) {
-                e.printStackTrace();
-            }
-            if(cnt%20 == 0){
-                syncTask();
-            }
+        }catch (Exception ex){
+            log.error(ex.getMessage());
+            System.out.println("出错");
         }
     }
 
@@ -113,6 +119,7 @@ public class DispatchTask implements Runnable {
         if(id > 0){
             // 是否加锁
             tasks.get(id).setTask_status(0);
+            System.out.println("任务" + id + "执行超时, 已重置");
         }
     }
     /**
@@ -142,6 +149,7 @@ public class DispatchTask implements Runnable {
             if(id > 0){
                 // 是否加锁
                 tasks.get(id).setTask_status(0);
+                System.out.println("任务" + id + "发送超时, 已重置");
             }
         }
     }
@@ -171,6 +179,8 @@ public class DispatchTask implements Runnable {
                 return null;
             clientIndex = (clientIndex + 1) % clientCount;
         }while(wrapper == null && clientIndex != 0);
+        if(wrapper != null)
+            System.out.println("当前选中index :" + clientIndex);
         return wrapper;
     }
 

+ 1 - 0
scheduler/src/main/java/cn/sagacloud/server/Server.java

@@ -91,6 +91,7 @@ public class Server {
 //            port = Integer.parseInt(args[0]);
 //        }
         new Server().start();
+
     }
 
 }

+ 1 - 2
scheduler/src/main/java/cn/sagacloud/utils/CommonUtil.java

@@ -32,8 +32,7 @@ public class CommonUtil {
         return bb.array();
     }
     public static long getTime() {
-        DateTimeFormatter format = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");
-        return Long.parseLong(LocalDateTime.now().format(format));
+        return System.currentTimeMillis()/1000;
     }
 
 }

+ 14 - 14
scheduler/src/main/proto/MessageProto.proto

@@ -1,17 +1,17 @@
 syntax = "proto3";
 
-package cn.sagacloud.proto;
-option optimize_for = SPEED;
-//option java_package = "cn.sagacloud.server.proto";
-option java_outer_classname = "MessageProto";
-//message Message{
-//    int32 cmd = 1;
-//    string className = 2;
-//    string content = 3;
-//}
+ package cn.sagacloud.proto;
+ option optimize_for = SPEED;
+ //option java_package = "cn.sagacloud.server.proto";
+ option java_outer_classname = "MessageProto";
+ //message Message{
+ //    int32 cmd = 1;
+ //    string className = 2;
+ //    string content = 3;
+ //}
 
-message Message{
-    string cmd = 1;
-    int32 taskId = 2;
-    string content = 3;
-}
+ message Message{
+     string cmd = 1;
+     int32 taskId = 2;
+     string content = 3;
+ }