Update AbstractCommandExecutor.java

将inReader.readLine()和logBuffer 分开，改成两个线程来处理。

dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java

@@ -83,6 +83,8 @@ public abstract class AbstractCommandExecutor {
      */
     protected final List<String> logBuffer;
     
+    protected boolean logOutputIsScuccess = false;
+
     /**
      * taskExecutionContext
      */
@@ -336,34 +338,49 @@ public abstract class AbstractCommandExecutor {
      */
     private void parseProcessOutput(Process process) {
         String threadLoggerInfoName = String.format(LoggerUtils.TASK_LOGGER_THREAD_NAME + "-%s", taskExecutionContext.getTaskAppId());
-        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
+        ExecutorService getOutputLogService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName + "-" + "getOutputLogService");
-        parseProcessOutputExecutorService.submit(new Runnable() {
+        getOutputLogService.submit(new Runnable() {
             @Override
             public void run() {
                 BufferedReader inReader = null;
 
                 try {
                     inReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
-                    String line = null;
+                    String line;
+                    while ((line = inReader.readLine()) != null) {
+                        if (line.startsWith("${setValue(")) {
+                            varPool.append(line.substring("${setValue(".length(), line.length() - 2));
+                            varPool.append("$VarPool$");
+                        } else {
+                            logBuffer.add(line);
+                        }
+                    }
+                } catch (Exception e) {
+                    logger.error(e.getMessage(), e);
+                } finally {
+                    logOutputIsScuccess = true;
+                    close(inReader);
+                }
+            }
+        });
 
+        ExecutorService parseProcessOutputExecutorService = ThreadUtils.newDaemonSingleThreadExecutor(threadLoggerInfoName);
+        parseProcessOutputExecutorService.submit(new Runnable() {
+            @Override
+            public void run() {
+                try {
                     long lastFlushTime = System.currentTimeMillis();
-                    while (logBuffer.size() > 0 || (line = inReader.readLine()) != null) {
+                    while (logBuffer.size() > 0 || !logOutputIsScuccess) {
-                        if (null != line) {
+                        if (logBuffer.size() > 0) {
-                            if (line.startsWith("${setValue(")) {
+                            lastFlushTime = flush(lastFlushTime);
-                                varPool.append(line.substring("${setValue(".length(), line.length() - 2));
+                        } else {
-                                varPool.append("$VarPool$");
+                            Thread.sleep(Constants.DEFAULT_LOG_FLUSH_INTERVAL);
-                            } else {
-                                logBuffer.add(line);
-                            }
-                            line = null;
                         }
-                        lastFlushTime = flush(lastFlushTime);
                     }
                 } catch (Exception e) {
                     logger.error(e.getMessage(), e);
                 } finally {
                     clear();
-                    close(inReader);
                 }
             }
         });