diff --git a/cutejava-front/src/utils/elementui/CsDialog.js b/cutejava-front/src/utils/elementui/CsDialog.js index 458fe0b4..0a709244 100644 --- a/cutejava-front/src/utils/elementui/CsDialog.js +++ b/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: '关闭' }) diff --git a/cutejava-front/src/utils/elementui/CsMessage.js b/cutejava-front/src/utils/elementui/CsMessage.js index aaed424f..68710f85 100644 --- a/cutejava-front/src/utils/elementui/CsMessage.js +++ b/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' }) diff --git a/cutejava-front/src/utils/elementui/CsNotification.js b/cutejava-front/src/utils/elementui/CsNotification.js index 4db9a5ab..0130a44b 100644 --- a/cutejava-front/src/utils/elementui/CsNotification.js +++ b/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 }) } } diff --git a/cutejava-front/src/views/componentsDemo/CutePipelineNodeDemo.vue b/cutejava-front/src/views/componentsDemo/CutePipelineNodeDemo.vue index ddba5443..2d4e5f15 100644 --- a/cutejava-front/src/views/componentsDemo/CutePipelineNodeDemo.vue +++ b/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 } } diff --git a/cutejava-front/src/views/home.vue b/cutejava-front/src/views/home.vue index b31643f8..4450ad7a 100644 --- a/cutejava-front/src/views/home.vue +++ b/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: { diff --git a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsClientManager.java b/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsClientManager.java index f516f686..6a73d3ff 100644 --- a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsClientManager.java +++ b/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 client = new ConcurrentHashMap<>(); + private static final Map 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 = client.get(sid); - if (wsServer != null) { + CsWsServer wsServer = clientMap.get(sid); + if (wsServer != null) { + try { +// log.info("关闭session, sid={}", sid); wsServer.getSession().close(); + + } catch (Exception e) { + // ignore } - } catch (Exception e) { - // ignore +// log.info("停止任务, sid={}", sid); + wsServer.stopTask(sid); +// log.info("移除客户端, sid={}", sid); + clientMap.remove(sid); } - client.remove(sid); } public static Collection getAllClient() { - return client.values(); + return clientMap.values(); } public static CsWsServer getClientBySid(String sid) { - return client.get(sid); + return clientMap.get(sid); } } diff --git a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsMessage.java b/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsMessage.java index 68e11693..509cda0a 100644 --- a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsMessage.java +++ b/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) { - this.bizType = bizType; + public CsWsMessage(String bizCode, String data) { + this.bizCode = bizCode; this.data = data; } } diff --git a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsServer.java b/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsServer.java index f0ab3dc4..8dd39fec 100644 --- a/cutejava/cutejava-framework/src/main/java/cn/odboy/framework/websocket/context/CsWsServer.java +++ b/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 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); - CsWsServer client = CsWsClientManager.getClientBySid(this.sid); - if (client != null) { - try { - client.getSession().close(); - } catch (IOException e) { - // ignore - } +// log.error("WebSocket sid={} 发生错误", this.sid, error); + try { 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 + } + } + } } diff --git a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/dal/model/DevOpsQueryLastPipelineDetailArgs.java b/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/dal/model/DevOpsQueryLastPipelineDetailArgs.java index a729f86c..03d728f6 100644 --- a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/dal/model/DevOpsQueryLastPipelineDetailArgs.java +++ b/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; } diff --git a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/framework/pipeline/core/PipelineJobBean.java b/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/framework/pipeline/core/PipelineJobBean.java index ba09a92d..d3c50ae7 100644 --- a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/framework/pipeline/core/PipelineJobBean.java +++ b/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()); - if (PipelineStatusEnum.SUCCESS.getCode().equals(pipelineInstanceNode.getCurrentNodeStatus())) { + // 检查实例节点状态 + ThreadUtil.safeSleep(3000); + 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); diff --git a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/PipelineInstanceService.java b/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/PipelineInstanceService.java index 93d6da3c..b3fc05ea 100644 --- a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/PipelineInstanceService.java +++ b/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); } diff --git a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/impl/PipelineInstanceServiceImpl.java b/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/impl/PipelineInstanceServiceImpl.java index 1e3d9ddb..edff2833 100644 --- a/cutejava/cutejava-module-devops/src/main/java/cn/odboy/devops/service/impl/PipelineInstanceServiceImpl.java +++ b/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 runningThreadMap = new HashMap<>(); @Override public StartPipelineResultVo startPipeline(PipelineInstanceTb pipelineInstanceTb) { @@ -109,32 +105,29 @@ public class PipelineInstanceServiceImpl implements PipelineInstanceService { } @Override - public void queryLastPipelineDetailWs(String instanceId) { - Thread thread = runningThreadMap.get(instanceId); - if (thread != null && !thread.isInterrupted()) { - try { - thread.stop(); - } catch (Exception e) { - // ignore - } + public void queryLastPipelineDetailWs(String sid) { + // sid {username}_{bizCode}_{contextParams} + String[] sids = sid.split("_"); + if (sids.length != 3) { + throw new BadRequestException("sid格式异常"); } - 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); - CsWsMessage message = new CsWsMessage("FetchPipelineLastDetail", JSON.toJSONString(pipelineInstanceVo)); - csWsServer.sendMessage(message, currentSid); - } catch (IOException e) { + PipelineInstanceVo pipelineInstanceVo = queryLastPipelineDetail(instanceId); + CsWsMessage message = new CsWsMessage(bizCode, JSON.toJSONString(pipelineInstanceVo)); + wsServer.sendMessage(message, sid); + } catch (Exception e) { log.error("推送流水线最新数据失败", e); loop = false; } } }); - thread.start(); - runningThreadMap.put(instanceId, thread); } } diff --git a/cutejava/cutejava-starter/src/main/resources/application-dev.yml b/cutejava/cutejava-starter/src/main/resources/application-dev.yml index 90ea1881..ded2adc3 100644 --- a/cutejava/cutejava-starter/src/main/resources/application-dev.yml +++ b/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: