[agent][executor]- bug - 15

agent/src/main/java/io/wdd/agent/executor/thread/CommandExecLogCache.java

@@ -3,6 +3,7 @@ package io.wdd.agent.executor.thread;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import io.wdd.agent.config.utils.AgentCommonThreadPool;
 import io.wdd.agent.executor.reply.SimpleStreamSender;
 import io.wdd.agent.message.OMessageToServerSender;
@@ -19,6 +20,9 @@ import java.io.IOException;
 import java.io.InputStreamReader;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
 
 
 /**
@@ -44,6 +48,22 @@ public class CommandExecLogCache {
      */
     private static final HashMap<String, InputStreamReader> CommandLogInputReaderMap = new HashMap<>();
 
+    /**
+     * 固定单进程,用于缓存命令执行日志,关闭命令输入管道
+     */
+    private static ExecutorService LogCacheDaemonThread = null;
+
+    static {
+
+        ThreadFactory threadFactory = new ThreadFactoryBuilder()
+                .setDaemon(true)
+                .setNameFormat("exec-log-cache")
+                .setPriority(7)
+                .build();
+
+        LogCacheDaemonThread = Executors.newSingleThreadExecutor(threadFactory);
+    }
+
     @Resource
     SimpleStreamSender simpleStreamSender;
 
@@ -137,6 +157,9 @@ public class CommandExecLogCache {
                 inputStreamReader
         );
 
+        // 使用统一进程执行缓存和关停动作
+        LogCacheDaemonThread.submit(
+                () -> {
                     log.debug("开始获取bufferedReader中每一行的内容");
                     // !! 此处会阻塞
                     // 阻塞读取命令执行日志的输出流
@@ -152,6 +175,8 @@ public class CommandExecLogCache {
                             commandCachedLog
                     );
                 }
+        );
+    }
 
     /**
      * 对于一些没有中止的任务,必须要手动将读取的 InputStream流关闭
@@ -169,6 +194,8 @@ public class CommandExecLogCache {
                 inputStreamReader
         );
 
+        LogCacheDaemonThread.submit(
+                () -> {
                     try {
                         // 关闭command的输出流
                         process
@@ -190,8 +217,7 @@ public class CommandExecLogCache {
                         CommandLogBufferedReaderMap.remove(streamKey);
                         CommandLogInputReaderMap.remove(streamKey);
                     }
-
+                });
-
     }
 
     /**