refactor(websocket): 重构 WebSocket 服务端

- 优化 CsWsClientManager 类,改用 clientMap 存储客户端连接
- 重构 CsWsMessage 类,将 bizType 改为 bizCode
- 优化 CsWsServer 类,增加任务线程管理功能
- 重构 PipelineInstanceService 接口,修改 queryLastPipelineDetailWs 方法参数
- 优化 PipelineInstanceServiceImpl 中的查询逻辑
- 更新前端 WebSocket 连接逻辑,采用新的 sid 格式

cutejava-front/src/utils/elementui/CsDialog.js

@@ -11,7 +11,7 @@ export default {
    * @param confirmCallBack 确认回调
    * @param cancelCallBack 取消回调
    */
-  FullConfirm(questionText, msgType, confirmCallBack, cancelCallBack) {
+  FullConfirm: function(questionText, msgType, confirmCallBack, cancelCallBack) {
     let currentType = 'info'
     if (msgType === 's') {
       currentType = 'success'
@@ -38,7 +38,7 @@ export default {
    * @param confirmCallBack 确认回调
    * @param cancelCallBack 取消回调
    */
-  DeleteConfirm(confirmCallBack, cancelCallBack) {
+  DeleteConfirm: function(confirmCallBack, cancelCallBack) {
     this.FullConfirm('确定删除本条数据吗？', 'w', confirmCallBack, cancelCallBack)
   }, /**
    * 数据删除确认对话框(自定义提示内容)
@@ -46,7 +46,7 @@ export default {
    * @param confirmCallBack 确认回调
    * @param cancelCallBack 取消回调
    */
-  DeleteMessageConfirm(message, confirmCallBack, cancelCallBack) {
+  DeleteMessageConfirm: function(message, confirmCallBack, cancelCallBack) {
     this.FullConfirm(message, 'w', confirmCallBack, cancelCallBack)
   }, /**
    * 数据输入对话框
@@ -56,7 +56,7 @@ export default {
    * @param confirmCallBack 校验通过回调
    * @param cancelCallBack 取消回调
    */
-  TextInputConfirm(title, patternText, errorMsg, confirmCallBack, cancelCallBack) {
+  TextInputConfirm: function(title, patternText, errorMsg, confirmCallBack, cancelCallBack) {
     Vue.prototype.$prompt('', title, {
       confirmButtonText: '确定', cancelButtonText: '取消', inputPattern: patternText, inputErrorMessage: errorMsg
     }).then(({ value }) => {
@@ -73,7 +73,7 @@ export default {
    * @param title 标题
    * @param content 内容
    */
-  ShowConfirm(title, content) {
+  ShowConfirm: function(title, content) {
     Vue.prototype.$alert(content, title, {
       confirmButtonText: '关闭'
     })

cutejava-front/src/utils/elementui/CsMessage.js

@@ -3,22 +3,22 @@ import Vue from 'vue'
  * 消息提示
  */
 export default {
-  Info(content) {
+  Info: function(content) {
     Vue.prototype.$message({
       message: content, type: 'info'
     })
   },
-  Success(content) {
+  Success: function(content) {
     Vue.prototype.$message({
       message: content, type: 'success'
     })
   },
-  Warning(content) {
+  Warning: function(content) {
     Vue.prototype.$message({
       message: content, type: 'warning'
     })
   },
-  Error(content) {
+  Error: function(content) {
     Vue.prototype.$message({
       message: content, type: 'error'
     })

cutejava-front/src/utils/elementui/CsNotification.js

@@ -4,17 +4,17 @@ import Vue from 'vue'
  * 通知
  */
 export default {
-  Info(content) {
+  Info: function(content) {
     Vue.prototype.$notify({ message: content, type: 'info' })
-  }, Success(content) {
+  }, Success: function(content) {
     Vue.prototype.$notify({ message: content, type: 'success' })
-  }, SuccessDuration(content, durationTime) {
+  }, SuccessDuration: function(content, durationTime) {
     Vue.prototype.$notify({ message: content, type: 'success', duration: durationTime })
-  }, Warning(content) {
+  }, Warning: function(content) {
     Vue.prototype.$notify({ message: content, type: 'warning' })
-  }, Error(content) {
+  }, Error: function(content) {
     Vue.prototype.$notify({ message: content, type: 'error' })
-  }, ErrorDuration(content, durationTime) {
+  }, ErrorDuration: function(content, durationTime) {
     Vue.prototype.$notify({ message: content, type: 'error', duration: durationTime })
   }
 }

cutejava-front/src/views/componentsDemo/CutePipelineNodeDemo.vue

@@ -71,6 +71,7 @@ import { queryLastPipelineDetail, queryLastPipelineDetailWs, startPipeline } fro
 import { CountArraysObjectByPropKey, FormatDateTimeStr } from '@/utils/CsUtil'
 import CsMessage from '@/utils/elementui/CsMessage'
 import CsWsClient from '@/utils/CsWsClient'
+import { mapGetters } from 'vuex'
 
 export default {
   name: 'CutePipelineNodeDemo',
@@ -472,6 +473,11 @@ export default {
     // }
     // this.connectWebSocketServer(pipelineInstanceId)
   },
+  computed: {
+    ...mapGetters([
+      'user'
+    ])
+  },
   methods: {
     refreshData() {
       // 成功
@@ -509,7 +515,8 @@ export default {
      */
     connectWebSocketServer(instanceId) {
       const that = this
-      const sid = 'instanceId_' + instanceId
+      // {username}_{bizCode}_{contextParams}
+      const sid = `${that.user.username}_FetchPipelineLastDetail_${instanceId}`
       that.dynamicWsClient = new CsWsClient(sid)
       that.dynamicWsClient.connect(that.handleWebSocketError, that.handleWebSocketMessage)
       setTimeout(() => {
@@ -533,11 +540,17 @@ export default {
         // 判断流水线是否结束
         const successCount = CountArraysObjectByPropKey(that.dynamicInstance.nodes, 'status', 'success')
         if (that.dynamicTemplate && that.dynamicTemplate.length === successCount) {
+          // 所有节点执行成功
           if (that.dynamicWsClient) {
             that.dynamicWsClient.close()
           }
           that.dynamicStartupStatus = that.dynamicStartupStatusMap.start.code
         } else {
+          const failCount = CountArraysObjectByPropKey(that.dynamicInstance.nodes, 'status', 'fail')
+          // 存在失败的节点
+          if (that.dynamicWsClient && failCount > 0) {
+            that.dynamicWsClient.close()
+          }
           that.dynamicStartupStatus = that.dynamicStartupStatusMap.restart.code
         }
       } catch (e) {
@@ -594,8 +607,9 @@ export default {
         CsMessage.Success('流水线启动成功')
         // await that.fetchLastDetail(result.instanceId)
         sessionStorage.setItem('pipelineInstanceId', result.instanceId)
-        this.connectWebSocketServer(result.instanceId)
+        that.connectWebSocketServer(result.instanceId)
       } catch (e) {
+        console.error('error', e)
         that.dynamicStartButtonLoading = false
       }
     }

cutejava-front/src/views/home.vue

@@ -36,7 +36,6 @@ import LineChart from './dashboard/LineChart'
 import RadarChart from '@/components/Echarts/RadarChart'
 import PieChart from '@/components/Echarts/PieChart'
 import BarChart from '@/components/Echarts/BarChart'
-import CsMessage from '@/utils/elementui/CsMessage'
 
 const lineChartData = {
   newVisitis: {

cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsClientManager.java

@@ -1,39 +1,52 @@
 package cn.odboy.framework.websocket.context;
 
-import javax.websocket.CloseReason;
+import lombok.extern.slf4j.Slf4j;
+
 import java.util.Collection;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
+/**
+ * WebSocket客户端管理
+ *
+ * @author odboy
+ * @date 2025-07-24
+ */
+@Slf4j
 public class CsWsClientManager {
     /**
      * concurrent包的线程安全Map, 用来存放每个客户端对应的MyWebSocket对象。（分布式必出问题）
      */
-    private static final Map<String, CsWsServer> client = new ConcurrentHashMap<>();
+    private static final Map<String, CsWsServer> clientMap = new ConcurrentHashMap<>();
 
     public static void addClient(String sid, CsWsServer csWsServer) {
         // 如果存在就先删除一个, 防止重复推送消息
         removeClient(sid);
-        client.put(sid, csWsServer);
+        clientMap.put(sid, csWsServer);
     }
 
     public static void removeClient(String sid) {
-        try {
+        CsWsServer wsServer = clientMap.get(sid);
-            CsWsServer wsServer = client.get(sid);
+        if (wsServer != null) {
-            if (wsServer != null) {
+            try {
+//                log.info("关闭session, sid={}", sid);
                 wsServer.getSession().close();
+
+            } catch (Exception e) {
+                // ignore
             }
-        } catch (Exception e) {
+//            log.info("停止任务, sid={}", sid);
-            // ignore
+            wsServer.stopTask(sid);
+//            log.info("移除客户端, sid={}", sid);
+            clientMap.remove(sid);
         }
-        client.remove(sid);
     }
 
     public static Collection<CsWsServer> getAllClient() {
-        return client.values();
+        return clientMap.values();
     }
 
     public static CsWsServer getClientBySid(String sid) {
-        return client.get(sid);
+        return clientMap.get(sid);
     }
 }

cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsMessage.java

@@ -7,11 +7,11 @@ import lombok.Setter;
 @Getter
 @Setter
 public class CsWsMessage extends CsObject {
-    private String bizType;
+    private String bizCode;
     private String data;
 
-    public CsWsMessage(String bizType, String data) {
+    public CsWsMessage(String bizCode, String data) {
-        this.bizType = bizType;
+        this.bizCode = bizCode;
         this.data = data;
     }
 }

cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsServer.java

@@ -9,7 +9,9 @@ import javax.websocket.*;
 import javax.websocket.server.PathParam;
 import javax.websocket.server.ServerEndpoint;
 import java.io.IOException;
+import java.util.Map;
 import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
 
 
 @Slf4j
@@ -17,6 +19,7 @@ import java.util.Objects;
 @ServerEndpoint("/websocket/{sid}")
 @Getter
 public class CsWsServer {
+    private static final Map<String, Thread> taskThreadMap = new ConcurrentHashMap<>();
     /**
      * 与某个客户端的连接会话, 需要通过它来给客户端发送数据
      */
@@ -32,10 +35,24 @@ public class CsWsServer {
     @OnOpen
     public void onOpen(Session session, @PathParam("sid") String sid) {
         this.session = session;
+        // {username}#{bizCode}#{contextParams}
         this.sid = sid;
         CsWsClientManager.addClient(sid, this);
     }
 
+    /**
+     * 收到客户端消息后调用的方法
+     *
+     * @param message 客户端发送过来的消息
+     */
+    @OnMessage
+    public void onMessage(String message, Session session) {
+        CsWsMessage wsMessage = JSON.parseObject(message, CsWsMessage.class);
+        String bizCode = wsMessage.getBizCode();
+        Object data = wsMessage.getData();
+        log.info("收到来 sid={} 的信息: message={}, bizCode={}, data={}", sid, message, bizCode, JSON.toJSONString(data));
+    }
+
     /**
      * 连接关闭调用的方法
      */
@@ -48,46 +65,13 @@ public class CsWsServer {
         }
     }
 
-    /**
-     * 收到客户端消息后调用的方法
-     *
-     * @param message 客户端发送过来的消息
-     */
-    @OnMessage
-    public void onMessage(String message, Session session) {
-        CsWsMessage wsMessage = JSON.parseObject(message, CsWsMessage.class);
-        String bizType = wsMessage.getBizType();
-        Object data = wsMessage.getData();
-        log.info("收到来 sid={} 的信息: message={}, bizType={}, data={}", sid, message, bizType, JSON.toJSONString(data));
-    }
-
-    /**
-     * 群发消息
-     *
-     * @param message /
-     */
-    private void sendToAll(String message) {
-        for (CsWsServer item : CsWsClientManager.getAllClient()) {
-            try {
-                item.innerSendMessage(message);
-            } catch (IOException e) {
-                log.error("发送消息给 sid={} 失败", item.sid, e);
-                CsWsClientManager.removeClient(item.sid);
-            }
-        }
-    }
-
     @OnError
     public void onError(Session session, Throwable error) {
-        log.error("WebSocket sid={} 发生错误", this.sid, error);
+//        log.error("WebSocket sid={} 发生错误", this.sid, error);
-        CsWsServer client = CsWsClientManager.getClientBySid(this.sid);
+        try {
-        if (client != null) {
-            try {
-                client.getSession().close();
-            } catch (IOException e) {
-                // ignore
-            }
             CsWsClientManager.removeClient(this.sid);
+        } catch (Exception e) {
+            // ignore
         }
     }
 
@@ -99,7 +83,7 @@ public class CsWsServer {
     }
 
     /**
-     * 群发自定义消息
+     * 精准推送消息
      */
     public void sendMessage(CsWsMessage message, @PathParam("sid") String sid) throws IOException {
         String body = JSON.toJSONString(message);
@@ -117,6 +101,20 @@ public class CsWsServer {
         }
     }
 
+    /**
+     * 群发消息
+     */
+    public void sendToAll(String message) {
+        for (CsWsServer item : CsWsClientManager.getAllClient()) {
+            try {
+                item.innerSendMessage(message);
+            } catch (IOException e) {
+                log.error("发送消息给 sid={} 失败", item.sid, e);
+                CsWsClientManager.removeClient(item.sid);
+            }
+        }
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -133,4 +131,34 @@ public class CsWsServer {
     public int hashCode() {
         return Objects.hash(session, sid);
     }
+
+    public Thread getTaskThread(String sid) {
+        Thread thread = taskThreadMap.get(sid);
+        if (thread == null) {
+            return null;
+        }
+        if (thread.isInterrupted()) {
+            taskThreadMap.remove(sid);
+            return null;
+        }
+        return thread;
+    }
+
+    public void restartTask(Runnable runnable) {
+        stopTask(this.sid);
+        Thread thread = new Thread(runnable);
+        thread.start();
+        taskThreadMap.put(this.sid, thread);
+    }
+
+    public void stopTask(String sid) {
+        Thread thread = getTaskThread(this.sid);
+        if (thread != null) {
+            try {
+                thread.stop();
+            } catch (Exception e) {
+                // ignore
+            }
+        }
+    }
 }

cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/dal/model/DevOpsQueryLastPipelineDetailArgs.java

@@ -3,11 +3,13 @@ package cn.odboy.devops.dal.model;
 import cn.odboy.base.CsObject;
 import lombok.Getter;
 import lombok.Setter;
+
+import javax.validation.constraints.NotBlank;
 import javax.validation.constraints.NotNull;
 
 @Getter
 @Setter
 public class DevOpsQueryLastPipelineDetailArgs extends CsObject {
-    @NotNull(message = "流水线实例Id必填")
+    @NotBlank(message = "流水线实例Id必填")
     private String instanceId;
 }

cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/framework/pipeline/core/PipelineJobBean.java

@@ -103,6 +103,7 @@ public class PipelineJobBean implements InterruptableJob {
                 }
 
                 try {
+                    // 流水线正式启动
                     pipelineNodeJobManage.startJob(pipelineInstanceTb, pipelineNodeTemplateVo);
                     pipelineInstanceTb.setCurrentNode(pipelineNodeTemplateVo.getCode());
                     pipelineInstanceTb.setCurrentNodeStatus(PipelineStatusEnum.RUNNING.getCode());
@@ -120,15 +121,16 @@ public class PipelineJobBean implements InterruptableJob {
                             pipelineInstanceMapper.updateById(pipelineInstanceTb);
                             break;
                         }
-                        ThreadUtil.safeSleep(5000);
+                        // 检查实例节点状态
-                        PipelineInstanceNodeTb pipelineInstanceNode = pipelineInstanceNodeService.getPipelineInstanceNodeByArgs(instanceId, pipelineNodeTemplateVo.getCode());
+                        ThreadUtil.safeSleep(3000);
-                        if (PipelineStatusEnum.SUCCESS.getCode().equals(pipelineInstanceNode.getCurrentNodeStatus())) {
+                        PipelineInstanceNodeTb runningInstanceNode = pipelineInstanceNodeService.getPipelineInstanceNodeByArgs(instanceId, pipelineNodeTemplateVo.getCode());
+                        if (PipelineStatusEnum.SUCCESS.getCode().equals(runningInstanceNode.getCurrentNodeStatus())) {
                             pipelineStatusEnum = PipelineStatusEnum.SUCCESS;
                             pipelineInstanceTb.setCurrentNodeStatus(pipelineStatusEnum.getCode());
                             pipelineInstanceMapper.updateById(pipelineInstanceTb);
                             break;
                         }
-                        if (PipelineStatusEnum.FAIL.getCode().equals(pipelineInstanceNode.getCurrentNodeStatus())) {
+                        if (PipelineStatusEnum.FAIL.getCode().equals(runningInstanceNode.getCurrentNodeStatus())) {
                             pipelineStatusEnum = PipelineStatusEnum.FAIL;
                             pipelineInstanceTb.setCurrentNodeStatus(pipelineStatusEnum.getCode());
                             pipelineInstanceMapper.updateById(pipelineInstanceTb);

cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/PipelineInstanceService.java

@@ -19,5 +19,5 @@ public interface PipelineInstanceService {
 
     PipelineInstanceVo queryLastPipelineDetail(String instanceId);
 
-    void queryLastPipelineDetailWs(String instanceId);
+    void queryLastPipelineDetailWs(String sid);
 }

cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/impl/PipelineInstanceServiceImpl.java

@@ -17,6 +17,7 @@ import cn.odboy.devops.service.PipelineInstanceNodeDetailService;
 import cn.odboy.devops.service.PipelineInstanceNodeService;
 import cn.odboy.devops.service.PipelineInstanceService;
 import cn.odboy.framework.exception.BadRequestException;
+import cn.odboy.framework.websocket.context.CsWsClientManager;
 import cn.odboy.framework.websocket.context.CsWsMessage;
 import cn.odboy.framework.websocket.context.CsWsServer;
 import com.alibaba.fastjson2.JSON;
@@ -24,11 +25,8 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
-import java.io.IOException;
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 
 @Slf4j
 @Service
@@ -39,8 +37,6 @@ public class PipelineInstanceServiceImpl implements PipelineInstanceService {
     private final PipelineInstanceDAO pipelineInstanceDAO;
     private final PipelineInstanceNodeService pipelineInstanceNodeService;
     private final PipelineInstanceNodeDetailService pipelineInstanceNodeDetailService;
-    private final CsWsServer csWsServer;
-    private volatile Map<String, Thread> runningThreadMap = new HashMap<>();
 
     @Override
     public StartPipelineResultVo startPipeline(PipelineInstanceTb pipelineInstanceTb) {
@@ -109,32 +105,29 @@ public class PipelineInstanceServiceImpl implements PipelineInstanceService {
     }
 
     @Override
-    public void queryLastPipelineDetailWs(String instanceId) {
+    public void queryLastPipelineDetailWs(String sid) {
-        Thread thread = runningThreadMap.get(instanceId);
+        // sid {username}_{bizCode}_{contextParams}
-        if (thread != null && !thread.isInterrupted()) {
+        String[] sids = sid.split("_");
-            try {
+        if (sids.length != 3) {
-                thread.stop();
+            throw new BadRequestException("sid格式异常");
-            } catch (Exception e) {
-                // ignore
-            }
         }
-        thread = new Thread(() -> {
+        CsWsServer wsServer = CsWsClientManager.getClientBySid(sid);
+        wsServer.restartTask(() -> {
+//            String username = sids[0];
+            String bizCode = sids[1];
+            String instanceId = sids[2];
             boolean loop = true;
-            final String currentSid = instanceId;
-            final String realInstanceId = currentSid.replace("instanceId_", "");
             while (loop) {
                 ThreadUtil.safeSleep(1000);
                 try {
-                    PipelineInstanceVo pipelineInstanceVo = queryLastPipelineDetail(realInstanceId);
+                    PipelineInstanceVo pipelineInstanceVo = queryLastPipelineDetail(instanceId);
-                    CsWsMessage message = new CsWsMessage("FetchPipelineLastDetail", JSON.toJSONString(pipelineInstanceVo));
+                    CsWsMessage message = new CsWsMessage(bizCode, JSON.toJSONString(pipelineInstanceVo));
-                    csWsServer.sendMessage(message, currentSid);
+                    wsServer.sendMessage(message, sid);
-                } catch (IOException e) {
+                } catch (Exception e) {
                     log.error("推送流水线最新数据失败", e);
                     loop = false;
                 }
             }
         });
-        thread.start();
-        runningThreadMap.put(instanceId, thread);
     }
 }

cutejava/cutejava-starter/src/main/resources/application-dev.yml

@@ -135,7 +135,7 @@ mybatis-plus:
     # 是否开启自动驼峰命名规则（camel case）映射, 即从经典数据库列名 A_COLUMN（下划线命名） 到经典 Java 属性名 aColumn（驼峰命名） 的类似映射
     map-underscore-to-camel-case: true
     # Sql日志
-    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
+#    log-impl: org.apache.ibatis.logging.stdout.StdOutImpl
   # 启动时是否检查 MyBatis XML 文件的存在, 默认不检查
   check-config-location: true
   # MyBatis Mapper 所对应的 XML 文件位置
@@ -236,7 +236,7 @@ mongo-plus:
       username: cutejava
       password: 123456
       authentication-database: cutejava
-  log: true
+  log: false
   pretty: false
   configuration:
     field: