[bug]增加内存不足等待机制,防止操作系统内存被耗光

datax-core/src/main/java/com/wugui/datatx/core/util/OsUtils.java

@@ -127,4 +127,9 @@ public static Boolean checkResource(double systemCpuLoad, double systemReservedM
         }
     }
 
+    public static int getAvailableMemorySize() {
+        SystemInfo systemInfo = new SystemInfo();
+        long available = systemInfo.getHardware().getMemory().getAvailable();
+        return (int) (available / (1024 * 1024));
+    }
 }

datax-executor/src/main/java/com/wugui/datax/executor/service/command/BuildCommand.java

@@ -1,5 +1,8 @@
 package com.wugui.datax.executor.service.command;
 
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.util.ReUtil;
+import cn.hutool.core.util.StrUtil;
 import com.wugui.datatx.core.biz.model.TriggerParam;
 import com.wugui.datatx.core.enums.IncrementTypeEnum;
 import com.wugui.datatx.core.util.Constants;
@@ -179,4 +182,23 @@ private static String buildPartition(List<String> partitionInfo) {
         return field + Constants.EQUAL + partitionTime;
     }
 
+    public static Integer insufficientResourcesWaiting(TriggerParam tgParam) {
+        int xmx = 1024;
+        if (StringUtils.isNotBlank(tgParam.getJvmParam())) {
+            String jvmParam = tgParam.getJvmParam().trim().toLowerCase();
+            if (StrUtil.containsAny(jvmParam, "-xmx")) {
+                String resultExtractMulti = ReUtil.extractMulti("\\-xmx.*?\\s", jvmParam, "$0");
+                resultExtractMulti = StrUtil.trim(StrUtil.removePrefix(resultExtractMulti, "-xmx"));
+                if (StrUtil.endWith(resultExtractMulti, "mb") || StrUtil.endWith(resultExtractMulti, "m")) {
+                    resultExtractMulti = StrUtil.removeSuffix(StrUtil.removeSuffix(resultExtractMulti, "mb"), "m");
+                    xmx = Convert.toInt(resultExtractMulti);
+                } else if (StrUtil.endWith(resultExtractMulti, "gb") || StrUtil.endWith(resultExtractMulti, "g")) {
+                    resultExtractMulti = StrUtil.removeSuffix(StrUtil.removeSuffix(resultExtractMulti, "gb"), "g");
+                    xmx = Convert.toInt(resultExtractMulti) * 1024;
+                }
+            }
+        }
+        return xmx;
+    }
+
 }

datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/DataXConstant.java

@@ -45,4 +45,6 @@ public class DataXConstant {
             "-queue {} " +
             "-datax_job {} " +
             "-datax_home_hdfs {} ";
+
+    public static final String INSUFFICIENT_MEMORY = "ip: {},insufficient memory, Need to memory: {}MB, The remaining memory: {}MB, waiting......";
 }

datax-executor/src/main/java/com/wugui/datax/executor/service/jobhandler/ExecutorJobHandler.java

@@ -2,19 +2,19 @@
 
 import cn.hutool.core.collection.CollUtil;
 import cn.hutool.core.io.FileUtil;
-import cn.hutool.core.util.IdUtil;
-import cn.hutool.core.util.ReUtil;
-import cn.hutool.core.util.RuntimeUtil;
-import cn.hutool.core.util.StrUtil;
+import cn.hutool.core.thread.ThreadUtil;
+import cn.hutool.core.util.*;
 import com.wugui.datatx.core.biz.model.HandleProcessCallbackParam;
 import com.wugui.datatx.core.biz.model.ReturnT;
 import com.wugui.datatx.core.biz.model.TriggerParam;
 import com.wugui.datatx.core.handler.AbstractJobHandler;
 import com.wugui.datatx.core.handler.annotation.JobHandler;
 import com.wugui.datatx.core.log.JobLogger;
 import com.wugui.datatx.core.thread.ProcessCallbackThread;
+import com.wugui.datatx.core.util.OsUtils;
 import com.wugui.datatx.core.util.PlaceHolder;
 import com.wugui.datatx.core.util.ProcessUtil;
+import com.wugui.datax.executor.service.command.BuildCommand;
 import com.wugui.datax.executor.service.logparse.LogStatistics;
 import com.wugui.datax.executor.util.SystemUtils;
 import org.apache.commons.lang3.StringUtils;
@@ -30,6 +30,7 @@
 import static com.wugui.datax.executor.service.command.BuildCommand.*;
 import static com.wugui.datax.executor.service.jobhandler.CheckEnv.checkEnv;
 import static com.wugui.datax.executor.service.jobhandler.DataXConstant.DEFAULT_JSON;
+import static com.wugui.datax.executor.service.jobhandler.DataXConstant.INSUFFICIENT_MEMORY;
 import static com.wugui.datax.executor.service.logparse.AnalysisStatistics.analysisStatisticsLog;
 
 /**
@@ -71,7 +72,7 @@ public class ExecutorJobHandler extends AbstractJobHandler {
 
     @Override
     public ReturnT<String> execute(TriggerParam trigger) {
-
+        insufficientResourcesWaiting(trigger);
         int exitValue = -1;
         Thread errThread = null;
         String tmpFilePath;
@@ -236,4 +237,20 @@ private String memory(String jvmParam,boolean unit){
         }
     }
 
+    /**
+     * 检查内存是否充足,不足时每五秒检查一次
+     *
+     * @param trigger
+     */
+    public void insufficientResourcesWaiting(TriggerParam trigger) {
+        int freePhysicalMemorySize = OsUtils.getAvailableMemorySize();
+        int memory = (int) (BuildCommand.insufficientResourcesWaiting(trigger) * 1.5);
+        while (freePhysicalMemorySize < memory) {
+            String logStr = StrUtil.format(INSUFFICIENT_MEMORY, NetUtil.getLocalhostStr(), memory, freePhysicalMemorySize);
+            JobLogger.log(logStr);
+            ThreadUtil.sleep(5000);
+            freePhysicalMemorySize = OsUtils.getAvailableMemorySize();
+        }
+    }
+
 }