modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/engine/manager/impl/MemoryLogManager.java

@@ -9,7 +9,6 @@ import jakarta.annotation.Resource;
 import jakarta.servlet.http.HttpServletRequest;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
-import org.springframework.util.CollectionUtils;
 import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;
 
 import java.io.IOException;
@@ -17,10 +16,6 @@ import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
 /**
  *  流水线运行时日志管理
  * */
@@ -30,12 +25,11 @@ public class MemoryLogManager implements LoggerManager {
     /**
      *  第一级 是taskId，可能同一个账号、IP、建立多个连接所以保存为list
      * */
-    private final  Map<String, List<SseEmitter>> taskIdSSEMap = new ConcurrentHashMap<>();
+    private final static Map<String, List<SseEmitter>> taskIdSSEMap = new ConcurrentHashMap<>();
 
-    private final  Map<String,StringBuffer> taskIdMemoryLogMap = new ConcurrentHashMap<>();
+    private final static Map<String,StringBuffer> taskIdMemoryLogMap = new ConcurrentHashMap<>();
 
-    public final Integer FLUSH_DB_SIZE=2*1024*1024;
+    public static final Integer FLUSH_DB_SIZE=2*1024*1024;
-    private ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
     /**
      *  缓存最近一次执行的日志，key是taskId，val是数据库id（已入库的情况下）用于buffer满了增加日志内容
      *  读取日志的时候同时读取数据库和内存中的日志
@@ -46,16 +40,15 @@ public class MemoryLogManager implements LoggerManager {
     @Override
     public SseEmitter subscribe(String taskId,HttpServletRequest request) {
         String ipAddr = IpUtil.getIpAddr(request);
-        SseEmitter emitter = new SseEmitter(30000L);
+        SseEmitter emitter = new SseEmitter(3000L);
         emitterInit(emitter,taskId);
         Long loginUserId = WebFrameworkUtils.getLoginUserId();
         log.info("SSE连接建立");
         log.info("当前请求ip：{}",ipAddr);
         log.info("当前用户id：{}",loginUserId);
         log.info("当前taskId：{}",loginUserId);
-        List<SseEmitter> taskIdSSEList = taskIdSSEMap.computeIfAbsent(taskId, k->new ArrayList<>(1));
+        List<SseEmitter> taskIdSSEList = taskIdSSEMap.getOrDefault(taskId, new ArrayList<>(1));
         taskIdSSEList.add(emitter);
-
         return emitter;
     }
 
@@ -66,7 +59,7 @@ public class MemoryLogManager implements LoggerManager {
          *  内存满4mb入库，查询日志和入库操作用同一把锁
          *  然后新内容推sse
          * */
-        StringBuffer logCache = taskIdMemoryLogMap.computeIfAbsent(taskId, k -> new StringBuffer());
+        StringBuffer logCache = taskIdMemoryLogMap.getOrDefault(taskId, new StringBuffer());
         int length = logCache.length();
         if (length>=FLUSH_DB_SIZE) {
             synchronized (this){
@@ -95,16 +88,6 @@ public class MemoryLogManager implements LoggerManager {
             }
         }
         logCache.append(logContent);
-        List<SseEmitter> sseEmitters = taskIdSSEMap.get(taskId);
-        if (!CollectionUtils.isEmpty(sseEmitters)) {
-            for (SseEmitter sseEmitter : sseEmitters) {
-                try {
-                    sseEmitter.send(logContent);
-                } catch (IOException e) {
-                    log.error("",e);
-                }
-            }
-        }
     }
 
     @Override
@@ -119,14 +102,6 @@ public class MemoryLogManager implements LoggerManager {
     }
 
     private void emitterInit(SseEmitter emitter,String taskId){
-        // 维持心跳
-        scheduler.scheduleWithFixedDelay(()-> {
-            try {
-                emitter.send("\n\n");
-            } catch (IOException e) {
-                log.error("",e);
-            }
-        },0,30,TimeUnit.SECONDS);
         emitter.onCompletion(()->{
             taskIdSSEMap.remove(taskId);
             log.info("===================taskId:{}断开连接===============",taskId);

modules/module-ci-process-biz/src/main/java/cd/casic/ci/process/process/service/task/impl/TaskServiceImpl.java

@@ -124,6 +124,7 @@ public class TaskServiceImpl extends ServiceImpl<PipTaskDao, PipTask> implements
 
     @Override
     public CommonResult<String> getLogContentByTaskId(String taskId) {
+
         return CommonResult.success(loggerManager.getLogContent(taskId));
     }
 }