执行日志封装

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

@@ -30,6 +30,69 @@
             <groupId>myth-job</groupId>
             <artifactId>myth-core-common</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>apache-log4j</groupId>
+            <artifactId>log4j</artifactId>
+            <version>1.2.8</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-simple</artifactId>
+            <version>1.7.25</version>
+            <scope>compile</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>2.6</version>
+        </dependency>
+        <dependency>
+            <groupId>joda-time</groupId>
+            <artifactId>joda-time</artifactId>
+            <version>2.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
     </dependencies>
 
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+                <configuration>
+                    <!-- 指定SpringBoot程序的main函数入口类 -->
+                    <mainClass>com.byit.executor.Test</mainClass>
+                </configuration>
+
+                <executions>
+                    <execution>
+                        <goals>
+                            <goal>repackage</goal>
+                        </goals>
+                    </execution>
+                </executions>
+            </plugin>
+
+            <plugin>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <configuration>
+                    <source>1.8</source>
+                    <target>1.8</target>
+                    <encoding>UTF-8</encoding>
+                    <compilerArguments>
+                        <!-- 打包本地jar包 -->
+                        <extdirs>${project.basedir}/lib</extdirs>
+                    </compilerArguments>
+                </configuration>
+            </plugin>
+        </plugins>
+    </build>
+
 </project>
\ No newline at end of file

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/Test2.java

+package com.byit.executor;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @description:
+ * @author: gml
+ * @create: 2020-01-15 17:28
+ */
+public class Test2 {
+
+    public static void main(String[] args) throws IOException, InterruptedException {
+        List<Process> processList = new ArrayList<>();
+        Thread thread = new Thread(() -> {
+            Runtime runtime = Runtime.getRuntime();
+            try {
+                Process exec = runtime.exec("python D:\\workspace\\pycharmWorkSpace\\test\\com.test\\Test.py");
+                processList.add(exec);
+                BufferedReader br = new BufferedReader(new InputStreamReader(exec.getInputStream()));
+                String res = null;
+                while ((res = br.readLine()) != null){
+                    System.out.println(res);
+                }
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        });
+
+        thread.start();
+
+        Thread.sleep(12000);
+
+        processList.forEach(process -> {
+            process.destroy();
+        });
+
+    }
+
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/MythJobProcess.java

+package com.byit.executor.jobExecutor.process;
+
+import com.byit.executor.util.LogGobbler;
+import com.google.common.base.Joiner;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @description: 封装进程类
+ * @author: gml
+ * @create: 2020-01-13 17:56
+ */
+@Slf4j
+public class MythJobProcess {
+    //杀死命令
+    public static String KILL_COMMAND = "kill";
+    //工作目录
+    private final String workingDir;
+    //命令
+    private final List<String> cmd;
+    //环境
+    private final Map<String, String> env;
+    //主线程的运行数
+    private final CountDownLatch startupLatch;
+    //获取日志的线程运行数
+    private final CountDownLatch completeLatch;
+    //进程id
+    private volatile int processId;
+    //执行命令的进程
+    private volatile Process process;
+    //以用户身份执行
+    private boolean isExecuteAsUser = false;
+    //作为用户二进制执行
+    private String executeAsUserBinary = null;
+    //有效用户
+    private String effectiveUser = null;
+
+    public MythJobProcess(final List<String> cmd, final Map<String, String> env,
+                          final String workingDir) {
+        this.cmd = cmd;
+        this.env = env;
+        this.workingDir = workingDir;
+        this.processId = -1;
+        this.startupLatch = new CountDownLatch(1);
+        this.completeLatch = new CountDownLatch(1);
+    }
+
+    public MythJobProcess(final List<String> cmd, final Map<String, String> env,
+                          final String workingDir, final String executeAsUserBinary,
+                          final String effectiveUser) {
+        this(cmd, env, workingDir);
+        this.isExecuteAsUser = true;
+        this.executeAsUserBinary = executeAsUserBinary;
+        this.effectiveUser = effectiveUser;
+    }
+
+    /**
+     * 执行此过程，直到完成为止
+     */
+    public String run() throws IOException {
+        //判断是否执行过
+        if (this.isStarted() || this.isComplete()) {
+            throw new IllegalStateException("该过程只能使用一次");
+        }
+
+        //标记为开始运行
+        this.startupLatch.countDown();
+
+        final ProcessBuilder builder = new ProcessBuilder(this.cmd);
+        if (!Objects.isNull(workingDir)){
+            builder.directory(new File(this.workingDir));
+        }
+        if (env != null && !env.isEmpty()){
+            builder.environment().putAll(this.env);
+        }
+        //重定向错误流
+        builder.redirectErrorStream(true);
+        //开始运行命令
+        this.process = builder.start();
+        LogGobbler outputGobbler = null;
+        LogGobbler errorGobbler = null;
+        try {
+            //获取进程的id
+            this.processId = processId(this.process);
+            if (this.processId == -1) {
+                log.info("未找到有效的进程id");
+            } else {
+                log.info("进程的id " + this.processId);
+            }
+
+            outputGobbler = new LogGobbler(new InputStreamReader(this.process.getInputStream(), StandardCharsets.UTF_8), Boolean.FALSE, 30);
+            errorGobbler = new LogGobbler(new InputStreamReader(this.process.getErrorStream(), StandardCharsets.UTF_8), Boolean.TRUE, 30);
+            //开始获取进程的运行日志
+            outputGobbler.start();
+            errorGobbler.start();
+            int exitCode = -1;
+            try {
+                //获取进程的运行状态
+                exitCode = this.process.waitFor();
+            } catch (final InterruptedException e) {
+                log.info("流程中断，退出状态为 " + exitCode, e);
+            }
+
+            //尝试等待所有内容退出后再退出
+            outputGobbler.awaitCompletion(5000);
+            errorGobbler.awaitCompletion(5000);
+
+            System.out.println(exitCode);
+            if (exitCode != 0) {
+                throw new ProcessFailureException(exitCode);
+            }
+
+        } finally {
+            //获取进程运行日志的状态完成
+            this.completeLatch.countDown();
+            this.process.getInputStream().close();
+            this.process.getOutputStream().close();
+            this.process.getErrorStream().close();
+            if (this.process.isAlive()){
+                this.process.destroy();
+            }
+            return outputGobbler.getRecentLog() + "\n" + errorGobbler.getRecentLog();
+        }
+    }
+
+    /**
+     * 等待进程执行完成
+     *
+     * @throws InterruptedException 如果线程在等待时被中断
+     */
+    public void awaitCompletion() throws InterruptedException {
+        this.completeLatch.await();
+    }
+
+    /**
+     * 等待进程执行完成
+     * <p>
+     * 当此方法返回时，作业过程已创建，并且this.processId已设置。
+     *
+     * @throws InterruptedException 如果线程在等待时被中断
+     */
+    public void awaitStartup() throws InterruptedException {
+        this.startupLatch.await();
+    }
+
+    /**
+     * 获取此进程的进程ID（如果已启动）
+     *
+     * @return 进程ID；如果无法获取，则为-1
+     */
+    public int getProcessId() {
+        checkStarted();
+        return this.processId;
+    }
+
+    /**
+     * 尝试终止该进程，等待特定时间使其消失
+     *
+     * @param time 等待时间
+     * @param unit 时间单位
+     * @return 如果此软终止将在给定的等待时间内终止该进程，则为true。
+     */
+    public boolean softKill(final long time, final TimeUnit unit) throws InterruptedException {
+        //判断是否已经开始运行
+        checkStarted();
+        if (isRunning()){
+            if (this.processId != -1) {
+                try {
+                    if (this.isExecuteAsUser) {
+                        final String cmd = String.format("%s %s %s -9 %d", this.executeAsUserBinary, this.effectiveUser, KILL_COMMAND, this.processId);
+                        System.out.println("执行命令"  + cmd);
+                        Runtime.getRuntime().exec(cmd);
+                    } else {
+                        final String cmd = String.format("%s -9 %d", KILL_COMMAND, this.processId);
+                        System.out.println("执行命令"  + cmd);
+                        Runtime.getRuntime().exec(cmd);
+                    }
+                    return this.completeLatch.await(time, unit);
+                } catch (final IOException e) {
+                    log.error("尝试杀死失败.", e);
+                }
+            }
+            return false;
+        }else {
+            throw new IllegalStateException("程序不在运行中");
+        }
+    }
+
+    /**
+     * 强制杀死这个过程
+     */
+    public void hardKill() {
+        //判断是否已经开始运行
+        checkStarted();
+        //判断是否正在执行
+        if (isRunning()) {
+            if (this.processId != -1) {
+                try {
+                    if (this.isExecuteAsUser) {
+                        final String cmd = String.format("%s %s %s -9 %d", this.executeAsUserBinary, this.effectiveUser, KILL_COMMAND, this.processId);
+                        System.out.println("执行命令"  + cmd);
+                        Runtime.getRuntime().exec(cmd);
+                    } else {
+                        final String cmd = String.format("%s -9 %d", KILL_COMMAND, this.processId);
+                        System.out.println("执行命令"  + cmd);
+                        Runtime.getRuntime().exec(cmd);
+                    }
+                } catch (final IOException e) {
+                    log.error("Kill attempt failed.", e);
+                }
+            }
+            this.process.destroy();
+        }else {
+            throw new IllegalStateException("程序不在运行中");
+        }
+    }
+
+    /**
+     * 尝试获取此进程的进程ID
+     *
+     * @param process 从中获取ID的过程
+     * @return 进程的ID
+     */
+    private int processId(final java.lang.Process process) {
+        int processId = -1;
+        try {
+            final Field f = process.getClass().getDeclaredField("pid");
+            f.setAccessible(true);
+
+            processId = f.getInt(process);
+        } catch (final Throwable e) {
+            log.error("获取进程id出错.{}", e.getMessage());
+        }
+        return processId;
+    }
+
+    /**
+     * @return 判断是否已经开始执行
+     */
+    public boolean isStarted() {
+        return this.startupLatch.getCount() == 0L;
+    }
+
+    /**
+     * @return 如果流程已完成，则为是
+     */
+    public boolean isComplete() {
+        return this.completeLatch.getCount() == 0L;
+    }
+
+    /**
+     * @return 如果进程当前正在运行，则为true
+     */
+    public boolean isRunning() {
+        return isStarted() && !isComplete();
+    }
+
+    public void checkStarted() {
+        if (!isStarted()) {
+            throw new IllegalStateException("程序尚未开始");
+        }
+    }
+
+    @Override
+    public String toString() {
+        return "Process(cmd = " + Joiner.on(" ").join(this.cmd) + ", env = " + this.env
+                + ", cwd = " + this.workingDir + ")";
+    }
+
+    public boolean isExecuteAsUser() {
+        return this.isExecuteAsUser;
+    }
+
+    public String getEffectiveUser() {
+        return this.effectiveUser;
+    }
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/MythJobProcessBuilder.java

+package com.byit.executor.jobExecutor.process;
+
+import org.apache.log4j.Logger;
+import com.google.common.base.Joiner;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @description:
+ * @author: gml
+ * @create: 2020-01-13 18:29
+ */
+public class MythJobProcessBuilder {
+    private final List<String> cmd = new ArrayList<>();
+    private Map<String, String> env = new HashMap<>();
+    private String workingDir = System.getProperty("user.dir");
+    private Logger logger = Logger.getLogger(MythJobProcess.class);
+    private boolean isExecuteAsUser = false;
+    private String executeAsUserBinaryPath = null;
+    private String effectiveUser = null;
+
+    private int stdErrSnippetSize = 30;
+    private int stdOutSnippetSize = 30;
+
+    public MythJobProcessBuilder(final String... command) {
+        addArg(command);
+    }
+
+    public MythJobProcessBuilder addArg(final String... command) {
+        for (final String c : command) {
+            this.cmd.add(c);
+        }
+        return this;
+    }
+
+    public MythJobProcessBuilder setWorkingDir(final String dir) {
+        this.workingDir = dir;
+        return this;
+    }
+
+    public String getWorkingDir() {
+        return this.workingDir;
+    }
+
+    public MythJobProcessBuilder setWorkingDir(final File f) {
+        return setWorkingDir(f.getAbsolutePath());
+    }
+
+    public MythJobProcessBuilder addEnv(final String variable, final String value) {
+        this.env.put(variable, value);
+        return this;
+    }
+
+    public Map<String, String> getEnv() {
+        return this.env;
+    }
+
+    public MythJobProcessBuilder setEnv(final Map<String, String> m) {
+        this.env = m;
+        return this;
+    }
+
+    public int getStdErrorSnippetSize() {
+        return this.stdErrSnippetSize;
+    }
+
+    public MythJobProcessBuilder setStdErrorSnippetSize(final int size) {
+        this.stdErrSnippetSize = size;
+        return this;
+    }
+
+    public int getStdOutSnippetSize() {
+        return this.stdOutSnippetSize;
+    }
+
+    public MythJobProcessBuilder setStdOutSnippetSize(final int size) {
+        this.stdOutSnippetSize = size;
+        return this;
+    }
+
+    public MythJobProcessBuilder setLogger(final Logger logger) {
+        this.logger = logger;
+        return this;
+    }
+
+    public MythJobProcess build() {
+        if (this.isExecuteAsUser) {
+            return new MythJobProcess(this.cmd, this.env, this.workingDir, this.executeAsUserBinaryPath, this.effectiveUser);
+        } else {
+            return new MythJobProcess(this.cmd, this.env, this.workingDir);
+        }
+    }
+
+    public List<String> getCommand() {
+        return this.cmd;
+    }
+
+    public String getCommandString() {
+        return Joiner.on(" ").join(getCommand());
+    }
+
+    @Override
+    public String toString() {
+        return "ProcessBuilder(cmd = " + Joiner.on(" ").join(this.cmd) + ", env = "
+                + this.env + ", cwd = " + this.workingDir + ")";
+    }
+
+    public MythJobProcessBuilder enableExecuteAsUser() {
+        this.isExecuteAsUser = true;
+        return this;
+    }
+
+    public MythJobProcessBuilder setExecuteAsUserBinaryPath(final String executeAsUserBinaryPath) {
+        this.executeAsUserBinaryPath = executeAsUserBinaryPath;
+        return this;
+    }
+
+    public MythJobProcessBuilder setEffectiveUser(final String effectiveUser) {
+        this.effectiveUser = effectiveUser;
+        return this;
+    }
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/jobExecutor/process/ProcessFailureException.java

+package com.byit.executor.jobExecutor.process;
+
+/**
+ * @description: 进程异常类
+ * @author: gml
+ * @create: 2020-01-13 17:56
+ */
+public class ProcessFailureException extends RuntimeException {
+
+  private static final long serialVersionUID = 1;
+
+  private final int exitCode;
+
+  public ProcessFailureException(final int exitCode) {
+    this.exitCode = exitCode;
+  }
+
+  public int getExitCode() {
+    return this.exitCode;
+  }
+
+  @Override
+  public String getMessage() {
+    return "Process exited with code " + this.exitCode;
+  }
+}

byit-myth-core/myth-executor-core/src/main/java/com/byit/executor/util/LogGobbler.java

+package com.byit.executor.util;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.Reader;
+
+/**
+ * @description: 获取运行日志的线程
+ * @author: gml
+ * @create: 2020-01-13 17:56
+ */
+@Slf4j
+public class LogGobbler extends Thread {
+
+    private final BufferedReader inputReader;
+    private final boolean isError;
+    private final StringBuffer buffer;
+
+    public LogGobbler(final Reader inputReader, final boolean isError, final int bufferLines) {
+        this.inputReader = new BufferedReader(inputReader);
+        this.isError = isError;
+        this.buffer = new StringBuffer(bufferLines);
+    }
+
+    @Override
+    public void run() {
+        try {
+            while (!Thread.currentThread().isInterrupted()) {
+                final String line = this.inputReader.readLine();
+                if (line == null) {
+                    return;
+                }
+
+                this.buffer.append(line + "\n");
+                printLog(line);
+            }
+        } catch (final IOException e) {
+            log.error("Error reading from logging stream:", e);
+        }
+    }
+
+    private void printLog(final String message) {
+        if (isError) {
+            log.error(message);
+        }else {
+            log.info(message);
+        }
+    }
+
+
+    /**
+     * 尝试等待所有内容退出后再退出
+     * @param waitMs
+     */
+    public void awaitCompletion(final long waitMs) {
+        try {
+            join(waitMs);
+        } catch (final InterruptedException e) {
+            log.error("I/O thread interrupted.", e);
+        }
+    }
+
+    public String getRecentLog() {
+        return buffer.toString();
+    }
+
+}

byit-myth-core/pom.xml

@@ -18,12 +18,17 @@
     </modules>
 
     <dependencies>
-    <dependency>
-        <groupId>org.projectlombok</groupId>
-        <artifactId>lombok</artifactId>
-    </dependency>
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
 
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+        </dependency>
 
-</dependencies>
+
+    </dependencies>
 
 </project>
\ No newline at end of file

pom.xml

@@ -64,8 +64,9 @@
 
     <mysql-connector-java.version>5.1.47</mysql-connector-java.version>
     <springfox-swagger2.version>2.9.2</springfox-swagger2.version>
-    <springfox-swagger-ui>2.9.2</springfox-swagger-ui>
-    <org-apache-commons>1.3</org-apache-commons>
+    <guava.version>21.0</guava.version>
+    <springfox-swagger-ui.version>2.9.2</springfox-swagger-ui.version>
+    <org-apache-commons.version>1.3</org-apache-commons.version>
 
 
     </properties>
@@ -76,7 +77,7 @@
             <dependency>
                 <groupId>org.apache.commons</groupId>
                 <artifactId>commons-exec</artifactId>
-                <version>${org-apache-commons}</version>
+                <version>${org-apache-commons.version}</version>
             </dependency>
             <dependency>
                 <groupId>org.springframework.cloud</groupId>
@@ -162,6 +163,12 @@
             </dependency>
 
             <dependency>
+                <groupId>com.google.guava</groupId>
+                <artifactId>guava</artifactId>
+                <version>${guava.version}</version>
+            </dependency>
+
+            <dependency>
                 <groupId>org.springframework.boot</groupId>
                 <artifactId>spring-boot-starter-mail</artifactId>
                 <version>${spring.boot.starter.mail.version}</version>
@@ -181,7 +188,7 @@
             <dependency>
                 <groupId>io.springfox</groupId>
                 <artifactId>springfox-swagger-ui</artifactId>
-                <version>${springfox-swagger-ui}</version>
+                <version>${springfox-swagger-ui.version}</version>
             </dependency>
 
         </dependencies>