Merge remote-tracking branch 'origin/developer' into developer

byit-myth-admin/src/main/java/com/byit/service/impl/ApiFlowServiceImpl.java

@@ -253,6 +253,7 @@ public class ApiFlowServiceImpl implements ApiFlowService {
             node.setPluginUrls(((PluginNode)pluginNode).getConfig().getPluginUrls());
             node.setRoutingStrategy(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getRoutingStrategy()) ? "RANDOM" : ((PluginNode)pluginNode).getConfig().getRoutingStrategy());
             node.setPriority(StringUtils.isEmpty(((PluginNode)pluginNode).getConfig().getPriority()) ? "1" : ((PluginNode)pluginNode).getConfig().getPriority());
+            node.setScriptUrls(((PluginNode)pluginNode).getScriptUrls());
             //设置失败重试
             if (null != ((PluginNode)pluginNode).getConfig().getFailedRetryCount()){
                 node.setFailedRetryCount(((PluginNode)pluginNode).getConfig().getFailedRetryCount());

byit-myth-core/myth-admin-core/src/main/java/com/byit/task/ScriptExecutorJobTask.java

@@ -50,6 +50,7 @@ public class ScriptExecutorJobTask implements TimerTask {
         scriptDto.setParam(mythJobTaskSchedule.getRunParam());
         scriptDto.setRunId(mythJobTaskSchedule.getRunId());
         scriptDto.setRemotePath(mythJobTaskSchedule.getScriptUrls());
+        scriptDto.setCallbackUrl("http://127.0.0.1:8998/job/callbackRes");
         DispatchResponseDto dispatchResponseDto = runScriptService.runScript(scriptDto);
         System.out.println(dispatchResponseDto);
     }

byit-myth-core/myth-core-common/src/main/java/com/byit/job/dto/ScriptDto.java

@@ -30,4 +30,8 @@ public class ScriptDto implements Serializable {
      * 脚本路径
      */
     private String remotePath;
+    /**
+     * 回调URL
+     */
+    private String callbackUrl;
 }

byit-myth-core/myth-executor-core/pom.xml

@@ -57,6 +57,10 @@
             <artifactId>junit</artifactId>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-autoconfigure</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/Test.java

-package com.byit.executor;
-
-import com.byit.executor.jobExecutor.process.MythJobProcess;
-import lombok.extern.slf4j.Slf4j;
-
-import java.io.IOException;
-import java.util.*;
-
-/**
- * @description:
- * @author: gml
- * @create: 2020-01-15 11:10
- */
-@Slf4j
-public class Test {
-
-    public static void main(String[] args) throws IOException, InterruptedException {
-
-        List<MythJobProcess> list = new ArrayList<>();
-
-        Thread thread = new Thread(() -> {
-            List<String> cmdList = new ArrayList<>();
-            cmdList = Arrays.asList("python D:\\workspace\\pycharmWorkSpace\\test\\com.test\\Test.py".split(" "));
-            Map<String, String> env = new HashMap<>();
-            env.put("python", "C:\\Program Files\\Python38");
-            MythJobProcess mythJobProcess = new MythJobProcess(cmdList, env, "D:\\workspace\\pycharmWorkSpace\\test\\com.test");
-            list.add(mythJobProcess);
-            try {
-                String log = mythJobProcess.run();
-                System.out.println("运行日志" + log);
-            } catch (IOException e) {
-                e.printStackTrace();
-            }
-        });
-
-        thread.start();
-
-
-        Thread.sleep(30000);
-
-        list.forEach(process -> {
-            process.hardKill();
-        });
-
-    }
-
-}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/filesystem/FastDfsFileSystem.java

@@ -4,6 +4,8 @@ import cn.hutool.core.collection.CollectionUtil;
 import org.csource.common.MyException;
 import org.csource.common.NameValuePair;
 import org.csource.fastdfs.*;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
+import org.springframework.stereotype.Component;
 
 
 import java.io.IOException;
@@ -15,16 +17,17 @@ import java.util.Set;
  * FastDFS设计模式
  * @author huangfu
  */
+@Component
+@ConditionalOnExpression("'${myth-job.filestystem}'.equals('FASTDFS')")
 public class FastDfsFileSystem implements FileSystem {
-    static {
-
+    public FastDfsFileSystem() {
         try {
             ClientGlobal.init("fdfs_client.conf");
         } catch (IOException | MyException e) {
             e.printStackTrace();
         }
-
     }
+
     @Override
     public String uploadFile(byte[] fileBuffer, String fileExtName, Map<String, String> mateDaTa) throws IOException, MyException {
         TrackerClient trackerClient = new TrackerClient();

byit-myth-executor/myth-executor-server/pom.xml

@@ -28,6 +28,17 @@
             <groupId>myth-job</groupId>
             <artifactId>byit-myth-rpc</artifactId>
         </dependency>
+
+       <!-- &lt;!&ndash; slf4j &ndash;&gt;
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-log4j12</artifactId>
+            <scope>test</scope>
+        </dependency>-->
     </dependencies>
 
 </project>
\ No newline at end of file

byit-myth-executor/myth-executor-server/src/main/java/com/byit/pool/RunThreadPool.java

+package com.byit.pool;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author huangfu
+ */
+public class RunThreadPool {
+    public static final ThreadPoolExecutor SCRIPT_RUN_THREAD_POOL = new ThreadPoolExecutor(
+            100,
+            400,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<Runnable>(600),
+            r ->new Thread(r, "Myth job script run thread-" + r.hashCode()));
+
+}

byit-myth-executor/myth-executor-server/src/main/java/com/byit/service/ScriptExecutorServiceImpl.java

 package com.byit.service;
 
+import com.alibaba.fastjson.JSON;
 import com.byit.executor.api.ScriptExecutorService;
+import com.byit.executor.filesystem.FileSystem;
 import com.byit.job.dto.DispatchResponseDto;
+import com.byit.job.dto.JobRunResultDto;
 import com.byit.job.dto.ScriptDto;
 import com.byit.job.enums.JobResultEnum;
+import com.byit.job.vo.ReturnResult;
+import com.byit.pool.RunThreadPool;
 import com.byit.rpc.remoting.provider.annotation.RpcService;
+import lombok.extern.slf4j.Slf4j;
+import org.csource.common.MyException;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import java.io.*;
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.Map;
+import java.util.UUID;
+
 /**
  * 脚本执行的实现
  * @author huangfu
  */
 @Service
 @RpcService
+@Slf4j
 public class ScriptExecutorServiceImpl implements ScriptExecutorService {
+
+    @Value("${myth-job.log.root.path}")
+    private String rootLogPath;
+    @Value("${myth-job.script.root.path}")
+    private String rootScriptPath;
+
+    private final FileSystem fileSystem;
+
+    public ScriptExecutorServiceImpl(FileSystem fileSystem) {
+        this.fileSystem = fileSystem;
+    }
+
     @Override
     public DispatchResponseDto runPythonScript(ScriptDto scriptDto) {
-        System.out.println("--------------3脚本调用开始-----------");
+        DispatchResponseDto dispatchResponseDto;
+        try {
+            RunThreadPool.SCRIPT_RUN_THREAD_POOL.execute(()->{
+                log.debug("--------------脚本调用开始-----------");
+                //创建回复对象
+                JobRunResultDto jobRunResultDto = new JobRunResultDto();
+                jobRunResultDto.setStartTime(new Date());
+                //设定运行标识
+                jobRunResultDto.setJobRunId(scriptDto.getRunId());
+                //获取回调通知URL
+                String callbackUrl = scriptDto.getCallbackUrl();
+
+                String remotePath = scriptDto.getRemotePath();
+                String command = scriptDto.getCommand();
+                String scriptPath = byteArrayToFile(remotePath);
+                command = command + " " + scriptPath;
+                log.info("-----------执行命令，{}-----------",command);
+                //设置结束时间
+                jobRunResultDto.setEndTime(new Date());
+                jobRunResultDto.setReturnResult(ReturnResult.SUCCESS);
+                jobRunResultDto.setLogId(scriptDto.getLogId());
+                cn.hutool.http.HttpUtil.post(callbackUrl, JSON.toJSONString(jobRunResultDto));
+                log.debug("--------------脚本调用结束-----------");
+            });
 
-        System.out.println("--------------3脚本调用结束-----------");
-        DispatchResponseDto dispatchResponseDto = DispatchResponseDto.builder()
-                .code(JobResultEnum.DISPATCH_SUCCESS.getCode())
-                .msg(JobResultEnum.DISPATCH_SUCCESS.getMsg())
-                .build();
+            dispatchResponseDto = DispatchResponseDto.builder()
+                    .code(JobResultEnum.DISPATCH_SUCCESS.getCode())
+                    .msg(JobResultEnum.DISPATCH_SUCCESS.getMsg())
+                    .build();
+        }catch (Exception e){
+            dispatchResponseDto = DispatchResponseDto.builder()
+                    .code(JobResultEnum.DISPATCH_FAIL.getCode())
+                    .msg(JobResultEnum.DISPATCH_FAIL.getMsg())
+                    .build();
+        }
         return dispatchResponseDto;
     }
+
+    /**
+     * 将脚本字节转换成文件
+     * @return 生成文件的本地路径
+     */
+    private String byteArrayToFile(String remotePath){
+
+        File rootPathMkdir = new File(rootScriptPath);
+        if(!rootPathMkdir.exists()){
+            rootPathMkdir.mkdirs();
+        }
+
+
+        OutputStream out = null;
+        File file = null;
+        try {
+            byte[] scriptByteArray = fileSystem.downloaderFile(remotePath);
+            Map<String, String> fileMate = fileSystem.getFileMate(remotePath);
+            String fileName = fileMate.get("filename");
+            fileName = UUID.randomUUID().toString().replace("-","")+fileName;
+            file = new File(rootPathMkdir,fileName);
+
+            if(!file.exists()){
+                file.createNewFile();
+            }
+
+            out = new FileOutputStream(file);
+            out.write(scriptByteArray);
+            return file.getPath();
+        } catch (IOException | MyException e) {
+            e.printStackTrace();
+        } finally {
+            if(out != null){
+                try {
+                    out.close();
+                } catch (IOException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+        return null;
+
+    }
+
+    /**
+     * 将字符串转换我数组
+     * @param logData
+     * @return
+     */
+    private byte[] stringToByteArray(String logData) {
+        return logData.getBytes(StandardCharsets.UTF_8);
+    }
+
 }

byit-myth-executor/myth-executor-server/src/main/resources/application-dev.properties

@@ -3,4 +3,11 @@ myth-rpc.remoting.port=7776
 myth-rpc.registry.address=http://localhost:8080/myth-register
 myth-rpc.registry.env=huangfu
 myth-rpc.registry.biz=byit-myth-job
-logging.config=classpath:logback.xml
\ No newline at end of file
+
+myth-job.filestystem=FASTDFS
+
+
+myth-job.log.root.path=/mythjob/log/
+myth-job.script.root.path=/mythjob/script/
+
+logging.config=classpath:logback.xml

demo-client/byit-demo-client/src/main/java/com/byit/job/ScriptAddFlow.java

@@ -70,15 +70,14 @@ public class ScriptAddFlow {
         pluginNode2.setType("node");
         pluginNode2.setAuthor("皇甫");
         pluginNode2.setJobType("SCRIPT");
-        pluginNode2.setHandlerName("addJob");
-        pluginNode2.setRunParam("add1");
+        pluginNode2.setScriptUrls("ddmp/M00/00/00/CgB4Al5PswCAGEeiAAAHa95jlv4490.log");
+        pluginNode2.setRunCommand("python");
+        pluginNode2.setRunParam("test1");
         pluginNodeConfig2.setFailedRetryCount(2);
         pluginNodeConfig2.setFailedRetryInterval(TimeUnit.MINUTES.toSeconds(2));
         pluginNodeConfig2.setNodeCron("0 0/7 * * * ? *");
         pluginNodeConfig2.setNodeTimeout(-1L);
-        pluginNodeConfig2.setPluginUrls("http://127.0.0.1:8888");
         pluginNodeConfig2.setPriority("1");
-        pluginNodeConfig2.setRoutingStrategy(LoadBalance.ROUND.name());
         pluginNode2.setConfig(pluginNodeConfig2);
         pluginNode2.setDependNodeNameList(Collections.singletonList("start"));