From b781ebfcec3b1f49a2bc892e90d7a29483ec4fe1 Mon Sep 17 00:00:00 2001 From: xiaohuo <14141624+dsgfhrh@user.noreply.gitee.com> Date: Fri, 24 Oct 2025 20:42:56 +0800 Subject: [PATCH] =?UTF-8?q?10.24=E7=AB=99=E5=86=85=E4=BF=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/message/messageController.java | 68 +++++++ ALOps_sys_backend/alops-framework/pom.xml | 1 + ALOps_sys_backend/alops-system/pom.xml | 17 ++ .../alops/system/config/WebSocketConfig.java | 20 ++ .../alops/system/domain/InternalMessage.java | 9 +- .../alops/system/hander/WebSocketServer.java | 153 ++++++++++++++ .../java/com/alops/system/job/GatherJob.java | 26 ++- .../system/mapper/InternalMessageMapper.java | 17 ++ .../system/service/IAlertSenderService.java | 2 +- .../service/IInternalMessageService.java | 74 +++++++ .../service/impl/AlertLaunchServiceImpl.java | 4 +- .../service/impl/AlertSenderServiceImpl.java | 74 +++++-- .../impl/InternalMessageServiceImpl.java | 123 +++++++++++ .../impl/ScheduleRulesServiceImpl.java | 192 ++++++++++++++---- .../mapper/system/InternalMessageMapper.xml | 37 +++- 15 files changed, 751 insertions(+), 66 deletions(-) create mode 100644 ALOps_sys_backend/alops-admin/src/main/java/com/alops/web/controller/message/messageController.java create mode 100644 ALOps_sys_backend/alops-system/src/main/java/com/alops/system/config/WebSocketConfig.java create mode 100644 ALOps_sys_backend/alops-system/src/main/java/com/alops/system/hander/WebSocketServer.java create mode 100644 ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IInternalMessageService.java create mode 100644 ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/InternalMessageServiceImpl.java diff --git a/ALOps_sys_backend/alops-admin/src/main/java/com/alops/web/controller/message/messageController.java b/ALOps_sys_backend/alops-admin/src/main/java/com/alops/web/controller/message/messageController.java new file mode 100644 index 00000000..d2642340 --- /dev/null +++ b/ALOps_sys_backend/alops-admin/src/main/java/com/alops/web/controller/message/messageController.java @@ -0,0 +1,68 @@ +package com.alops.web.controller.message; + +import com.alops.common.annotation.Log; +import com.alops.common.core.domain.AjaxResult; +import com.alops.common.enums.BusinessType; +import com.alops.common.utils.SecurityUtils; +import com.alops.system.domain.InternalMessage; +import com.alops.system.domain.dto.AlertRuleAddDto; +import com.alops.system.domain.vo.AlertRuleVo; +import com.alops.system.service.IInternalMessageService; +import io.swagger.annotations.ApiOperation; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.security.access.prepost.PreAuthorize; +import org.springframework.web.bind.annotation.*; + +import java.util.List; + +@RestController +@RequestMapping("/message") +public class messageController { + + @Autowired + public IInternalMessageService iInternalMessageService; + + //变已读 + @PreAuthorize("@ss.hasPermi('message:change')") + @ApiOperation("变已读") + @Log(title = "【变已读】") + @PostMapping("/change/{ids}") + public AjaxResult change(@PathVariable Long[] ids) { + return AjaxResult.success("修改成功", iInternalMessageService.markMessagesAsRead(ids)); + + } + + //根据用户id获取未读量 + @PreAuthorize("@ss.hasPermi('message:read')") + @ApiOperation("未读量") + @Log(title = "【未读量】") + @PostMapping("/read") + public AjaxResult read() { + Long userId = SecurityUtils.getLoginUser().getUserId(); + + // 调用Service + int count = iInternalMessageService.getUnreadCountByUserId(userId); + + return AjaxResult.success("查询成功", count); + + } + + @PreAuthorize("@ss.hasPermi('message:view')") + @ApiOperation("获取用户消息列表") + @Log(title = "【消息列表】") + @GetMapping("/list") + public AjaxResult getMessageList() { + // 获取当前登录用户ID + Long userId = SecurityUtils.getLoginUser().getUserId(); + + // 调用Service获取消息列表 + List messages = iInternalMessageService.getMessagesByUserId(userId); + + return AjaxResult.success(messages); + } + + + + + +} diff --git a/ALOps_sys_backend/alops-framework/pom.xml b/ALOps_sys_backend/alops-framework/pom.xml index 03d53379..d8501072 100644 --- a/ALOps_sys_backend/alops-framework/pom.xml +++ b/ALOps_sys_backend/alops-framework/pom.xml @@ -60,6 +60,7 @@ com.alops alops-system + org.snmp4j snmp4j diff --git a/ALOps_sys_backend/alops-system/pom.xml b/ALOps_sys_backend/alops-system/pom.xml index 7ac7efa4..664ee492 100644 --- a/ALOps_sys_backend/alops-system/pom.xml +++ b/ALOps_sys_backend/alops-system/pom.xml @@ -60,6 +60,19 @@ + + + + org.springframework.boot + spring-boot-starter-websocket + + + + com.fasterxml.jackson.core + jackson-databind + + + org.quartz-scheduler quartz @@ -79,6 +92,10 @@ lombok provided + + org.springframework.boot + spring-boot-starter-websocket + diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/config/WebSocketConfig.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/config/WebSocketConfig.java new file mode 100644 index 00000000..85956d61 --- /dev/null +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/config/WebSocketConfig.java @@ -0,0 +1,20 @@ +package com.alops.system.config; + +import com.alops.system.hander.WebSocketServer; +import org.springframework.context.annotation.Configuration; +import org.springframework.web.socket.config.annotation.EnableWebSocket; +import org.springframework.web.socket.config.annotation.WebSocketConfigurer; +import org.springframework.web.socket.config.annotation.WebSocketHandlerRegistry; + +//理解为WebSocket的controller +@Configuration +@EnableWebSocket +public class WebSocketConfig implements WebSocketConfigurer { + @Override + public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) { + + registry.addHandler(new WebSocketServer(), "/ws/onlineMessage") + .setAllowedOrigins("*"); + + } +} \ No newline at end of file diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/domain/InternalMessage.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/domain/InternalMessage.java index 823e6d1f..ebe2a431 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/domain/InternalMessage.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/domain/InternalMessage.java @@ -6,6 +6,8 @@ import lombok.Data; import org.apache.commons.lang3.builder.ToStringBuilder; import org.apache.commons.lang3.builder.ToStringStyle; +import java.util.Date; + /** * 站内信对象 internal_message * @@ -13,7 +15,7 @@ import org.apache.commons.lang3.builder.ToStringStyle; * @date 2025-09-04 */ @Data -public class InternalMessage extends BaseEntity +public class InternalMessage { private static final long serialVersionUID = 1L; @@ -32,7 +34,11 @@ public class InternalMessage extends BaseEntity @Excel(name = "消息状态(unread/read)") private String status; + @Excel(name = "主题") + private String title; + @Excel(name = "创建时间") + private Date createTime; public void setId(Long id) { @@ -82,6 +88,7 @@ public class InternalMessage extends BaseEntity .append("content", getContent()) .append("status", getStatus()) .append("createTime", getCreateTime()) + .append("title", getTitle()) .toString(); } } diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/hander/WebSocketServer.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/hander/WebSocketServer.java new file mode 100644 index 00000000..db358635 --- /dev/null +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/hander/WebSocketServer.java @@ -0,0 +1,153 @@ +package com.alops.system.hander; + +import com.fasterxml.jackson.databind.ObjectMapper; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Component; +import org.springframework.util.MultiValueMap; +import org.springframework.web.socket.CloseStatus; +import org.springframework.web.socket.TextMessage; +import org.springframework.web.socket.WebSocketSession; +import org.springframework.web.socket.handler.TextWebSocketHandler; +import org.springframework.web.util.UriComponentsBuilder; +import com.fasterxml.jackson.databind.ObjectMapper; + +import java.util.Date; +import java.util.HashMap; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; + +@Component +public class WebSocketServer extends TextWebSocketHandler { + private static final Logger logger = LoggerFactory.getLogger(WebSocketServer.class); + // 存储在线用户 (Key: 用户ID, Value: WebSocketSession) + public static final ConcurrentHashMap onlineUsers = new ConcurrentHashMap<>(); + + + private static ObjectMapper objectMapper; + + + @Autowired + public void setObjectMapper(ObjectMapper mapper) { + WebSocketServer.objectMapper = mapper; + } + + @Override + public void afterConnectionEstablished(WebSocketSession session) throws Exception { + Long userId = getUserIdFromSession(session); + if (userId != null) { + logger.info("WebSocket 连接成功，用户上线: " + userId); + onlineUsers.put(userId, session); + } else { + logger.warn("无效的用户连接，关闭连接: " + session.getId()); + session.close(); // 没有 userId 的连接直接关闭 + } + } + + @Override + public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception { + Long userId = getUserIdFromSession(session); + if (userId != null) { + + logger.info("WebSocket 连接关闭，用户下线: " + userId + ", 关闭状态: " + status); + onlineUsers.remove(userId); + + } + } + + // 获取 WebSocketSession 中的 userId，并转换为 Long 类型 + public static Long getUserIdFromSession(WebSocketSession session) { + String query = session.getUri().getQuery(); + logger.info("WebSocket连接查询参数: {}", query); + + if (query != null) { + MultiValueMap params = UriComponentsBuilder.newInstance() + .query(query) + .build() + .getQueryParams(); + String userIdStr = params.getFirst("userId"); + if (userIdStr != null) { + try { + return Long.parseLong(userIdStr); + } catch (NumberFormatException e) { + logger.error("WebSocket连接无效的用户ID: {}", userIdStr, e); + } + } else { + logger.warn("WebSocket连接没有 userId 参数"); + } + } else { + logger.warn("WebSocket连接没有查询参数"); + } + return null; + } +// +// // 发送消息给指定用户 +// public void sendMessageToUser(Long userId, String content, String title) throws Exception { +// // 获取对应用户的 WebSocket session +// WebSocketSession session = onlineUsers.get(userId); +// +// // 检查 session 是否有效且连接未关闭 +// if (session != null && session.isOpen()) { +// // 创建一个 Map 来构建消息内容 +// Map messageMap = new HashMap<>(); +// messageMap.put("from UserName", fromUserName); // 发送者姓名 +// messageMap.put("content", content); // 消息内容 +// +// // 使用 Jackson ObjectMapper 将 Map 转换为 JSON 字符串 +// ObjectMapper objectMapper = new ObjectMapper(); +// String jsonMessage = objectMapper.writeValueAsString(messageMap); +// +// // 发送 JSON 格式的消息 +// session.sendMessage(new TextMessage(jsonMessage)); +// } +// } + + + /** + * 发送通知到指定用户 + * @param messageId 数据库中站内信的ID + * @param title 通知标题，例如 "设备异常预警" + * @param content 通知内容 + */ + public static void sendAlertMessage(Long userId, Long messageId, String title, String content) { + WebSocketSession session = onlineUsers.get(userId); + try { + if (session != null && session.isOpen()) { + + // 构建消息内容 + Map messageMap = new HashMap<>(); + messageMap.put("messageId", messageId); // 对应数据库ID + messageMap.put("title", title); // 弹窗标题 + messageMap.put("content", content); // 弹窗内容 + messageMap.put("timestamp", new Date()); // 发送时间 + + // 转换为 JSON + String jsonMessage = objectMapper.writeValueAsString(messageMap); + + // 发送 + session.sendMessage(new TextMessage(jsonMessage)); + logger.info("成功向用户 [{}] 发送预警消息: {}", session.getId(), jsonMessage); + } + } catch (Exception e) { + logger.error("发送预警消息失败: messageId=" + messageId, e); + } + } + + + public void sendMessageToUser1(Long userId, String message) throws Exception { + WebSocketSession session = onlineUsers.get(userId); + if (session != null && session.isOpen()) { + session.sendMessage(new TextMessage(message)); + } + } + + // 获取所有在线用户 ID (Long 类型) + public Set getOnlineUserIds() { + return onlineUsers.keySet(); + } + + + +} diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java index d7a376d6..dc3dffa5 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/job/GatherJob.java @@ -43,12 +43,15 @@ public class GatherJob implements Job { if (rule.getStatus() == 4) { Scheduler scheduler = context.getScheduler(); JobKey jobKey = context.getJobDetail().getKey(); - scheduler.deleteJob(jobKey); - String Content=String.format("任务执行完此次，已取消后续任务"); - - iAlertSenderService.sendInternalMessage( Content,"2" ); - log.info("规则[{}] 已标记为取消状态(4)，对应定时任务已删除: {}", rule.getId(), jobKey); - +// scheduler.deleteJob(jobKey); + boolean deleted = scheduler.deleteJob(jobKey); + if (deleted) { + iAlertSenderService.sendInternalMessage("采集任务取消成功","任务执行完此次，已取消后续任务","2"); + log.info("规则[{}] 已标记为取消状态(4)，对应定时任务已删除: {}", rule.getId(), jobKey); + } else { + log.warn("规则[{}] 尝试取消定时任务失败: {}", rule.getId(), jobKey); + iAlertSenderService.sendInternalMessage("采集任务取消失败","尝试取消定时任务失败，请检查调度器状态","2"); + } return; } // 2.设置状态为执行中 @@ -64,6 +67,8 @@ public class GatherJob implements Job { if ( now.after(end)) { rule.setStatus(2); // 超出范围标记完成 scheduleRulesMapper.updateScheduleRules(rule); + iAlertSenderService.sendInternalMessage("采集任务成功","按照频率进行采集的任务执行完成","2"); + return; } } @@ -76,10 +81,10 @@ public class GatherJob implements Job { if (rule.getRuleType() == 1 ) { rule.setStatus(2); - String Content=String.format("任务执行完成", + String Content=String.format("定时采集任务执行完成", rule.getId(), rule.getRuleName()); - iAlertSenderService.sendInternalMessage( Content,"2" ); + iAlertSenderService.sendInternalMessage("采集任务成功" ,Content,"2" ); scheduleRulesMapper.updateScheduleRules(rule); } @@ -95,10 +100,11 @@ public class GatherJob implements Job { rule.setStatus(3); scheduleRulesMapper.updateScheduleRules(rule); try { - String Content=String.format("任务执行失败，规则ID=%s，规则名称=%s，原因：%s", + String Content=String.format("采集任务执行失败，规则ID=%s，规则名称=%s，原因：%s", rule.getId(), rule.getRuleName(), e.getMessage()); - iAlertSenderService.sendInternalMessage( Content,"2" ); + iAlertSenderService.sendInternalMessage("采集任务失败" ,Content,"2" ); + log.info("已发送失败通知给运维人员"); } catch (Exception ne) { diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/mapper/InternalMessageMapper.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/mapper/InternalMessageMapper.java index f7cef6e8..b845f2cb 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/mapper/InternalMessageMapper.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/mapper/InternalMessageMapper.java @@ -15,6 +15,23 @@ import java.util.List; @Mapper public interface InternalMessageMapper { + + + /** + * 查询指定用户的所有消息，按时间倒序 + */ + List selectMessagesByUserId(@Param("userId") Long userId); + /** + * 统计某个用户的未读消息数量 + */ + int countUnreadByUserId(@Param("userId") Long userId); + + /** + * 批量更新消息状态为已读 + * @param ids 消息ID数组 + * @return 更新行数 + */ + int updateMessageStatusToRead( Long[] ids); /** * 批量插入消息 * @param messages 消息列表 diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IAlertSenderService.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IAlertSenderService.java index 5552dbbc..831526ec 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IAlertSenderService.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IAlertSenderService.java @@ -4,5 +4,5 @@ import java.util.List; public interface IAlertSenderService { public void sendAlert(List alertMessage, String alertWay, String target); - public void sendInternalMessage(String content, String roleIdStr); + public void sendInternalMessage(String title, String content, String roleIdStr); } diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IInternalMessageService.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IInternalMessageService.java new file mode 100644 index 00000000..82dc9284 --- /dev/null +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/IInternalMessageService.java @@ -0,0 +1,74 @@ +package com.alops.system.service; + +import java.util.List; + +import com.alops.system.domain.InternalMessage; + + +/** + * 站内信Service接口 + * + * @author ruoyi + * @date 2025-10-23 + */ +public interface IInternalMessageService +{ + + + /** + * 根据用户ID获取消息列表，按时间排序 + */ + List getMessagesByUserId(Long userId); + /** + * 根据用户ID获取未读消息数量 + */ + int getUnreadCountByUserId(Long userId); + public int markMessagesAsRead(Long[] ids); + /** + * 查询站内信 + * + * @param id 站内信主键 + * @return 站内信 + */ + public InternalMessage selectInternalMessageById(Long id); + + /** + * 查询站内信列表 + * + * @param internalMessage 站内信 + * @return 站内信集合 + */ + public List selectInternalMessageList(InternalMessage internalMessage); + + /** + * 新增站内信 + * + * @param internalMessage 站内信 + * @return 结果 + */ + public int insertInternalMessage(InternalMessage internalMessage); + + /** + * 修改站内信 + * + * @param internalMessage 站内信 + * @return 结果 + */ + public int updateInternalMessage(InternalMessage internalMessage); + + /** + * 批量删除站内信 + * + * @param ids 需要删除的站内信主键集合 + * @return 结果 + */ + public int deleteInternalMessageByIds(Long[] ids); + + /** + * 删除站内信信息 + * + * @param id 站内信主键 + * @return 结果 + */ + public int deleteInternalMessageById(Long id); +} diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertLaunchServiceImpl.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertLaunchServiceImpl.java index d297b6ce..aa52820e 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertLaunchServiceImpl.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertLaunchServiceImpl.java @@ -113,7 +113,7 @@ public class AlertLaunchServiceImpl implements IAlertLaunchService { * 同步插入消息 */ @Transactional - public void sendInternalMessages(List messages) { + public void insertAlertMessages(List messages) { if (messages == null || messages.isEmpty()) { return; } @@ -165,7 +165,7 @@ public class AlertLaunchServiceImpl implements IAlertLaunchService { if (!messages.isEmpty()) { alertSender.sendAlert(msgs, rule.getAlertWay(), rule.getAlertStation()); - sendInternalMessages(messages); + insertAlertMessages(messages); } } catch (Exception e) { diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertSenderServiceImpl.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertSenderServiceImpl.java index 9039fbd5..bc0f7243 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertSenderServiceImpl.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/AlertSenderServiceImpl.java @@ -3,10 +3,13 @@ package com.alops.system.service.impl; import com.alops.common.core.domain.entity.SysUser; import com.alops.system.domain.AlertMessage; import com.alops.system.domain.InternalMessage; +import com.alops.system.hander.WebSocketServer; import com.alops.system.mapper.InternalMessageMapper; import com.alops.system.mapper.SysUserMapper; import com.alops.system.service.IAlertSenderService; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.mail.SimpleMailMessage; @@ -15,9 +18,7 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; -import java.util.ArrayList; -import java.util.Date; -import java.util.List; +import java.util.*; @Service public class AlertSenderServiceImpl implements IAlertSenderService { @@ -33,6 +34,8 @@ public class AlertSenderServiceImpl implements IAlertSenderService { @Value("${spring.mail.username}") private String from; + private static final Logger log = LoggerFactory.getLogger(AlertSenderServiceImpl.class); + /** * 发送预警消息 * @@ -45,8 +48,8 @@ public class AlertSenderServiceImpl implements IAlertSenderService { // 将多条消息合并成一条 String combinedMessage = String.join("\n", alertMessages); - //运维人员2 - sendInternalMessage(combinedMessage,"2" ); + //运维人员2 + sendInternalMessage("预警信息",combinedMessage, "2"); if ("email".equalsIgnoreCase(alertWay)) { sendEmail(combinedMessage, target); @@ -57,8 +60,6 @@ public class AlertSenderServiceImpl implements IAlertSenderService { } - - /** * 邮件发送 */ @@ -81,11 +82,33 @@ public class AlertSenderServiceImpl implements IAlertSenderService { /** * 站内信发送 */ - public void sendInternalMessage(String content, String roleIdStr) { +// public void sendInternalMessage(String content, String roleIdStr) { +// long roleId = Long.parseLong(roleIdStr); +// List users = userMapper.selectByRoleId(roleId); +// if (users == null || users.isEmpty()) return; +// //1.插入站内信表 +// List messages = new ArrayList<>(); +// for (SysUser user : users) { +// InternalMessage msg = new InternalMessage(); +// msg.setUserId(user.getUserId()); +// msg.setContent(content); // 一次插入整条合并后的消息 +// msg.setCreateTime(new Date()); +// msg.setStatus("unread"); +// messages.add(msg); +// } +// if (!messages.isEmpty()) { +// insertMessages(messages); +// } +// //2.发送实时消息 +// +// +// } + + public void sendInternalMessage(String title, String content, String roleIdStr) { long roleId = Long.parseLong(roleIdStr); List users = userMapper.selectByRoleId(roleId); if (users == null || users.isEmpty()) return; - + //1.插入站内信表 List messages = new ArrayList<>(); for (SysUser user : users) { InternalMessage msg = new InternalMessage(); @@ -93,23 +116,46 @@ public class AlertSenderServiceImpl implements IAlertSenderService { msg.setContent(content); // 一次插入整条合并后的消息 msg.setCreateTime(new Date()); msg.setStatus("unread"); + msg.setTitle(title); messages.add(msg); } + List> result = insertMessages(messages); + + //2.发送实时消息 + for (Map item : result) { + Long userId = (Long) item.get("userId"); + Long messageId = (Long) item.get("id"); - if (!messages.isEmpty()) { - insertMessages(messages); + // 调用 WebSocket 工具发送 + WebSocketServer.sendAlertMessage(userId, messageId, title, content); } + + log.info("共发送站内信 {} 条（角色ID={}）", result.size(), roleId); } + @Transactional(rollbackFor = Exception.class) - public void insertMessages(List messages) { + public List> insertMessages(List messages) { if (messages == null || messages.isEmpty()) { - return; + return Collections.emptyList(); } - int rows = messageMapper.batchInsert(messages); // 返回插入行数 + + int rows = messageMapper.batchInsert(messages); if (rows != messages.size()) { throw new RuntimeException("部分消息未插入成功，事务回滚"); } + + // 组装返回结果 + List> resultList = new ArrayList<>(); + for (InternalMessage msg : messages) { + Map map = new HashMap<>(); + map.put("id", msg.getId()); // 主键已被MyBatis自动回填 + map.put("userId", msg.getUserId()); + resultList.add(map); + } + + return resultList; } } + diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/InternalMessageServiceImpl.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/InternalMessageServiceImpl.java new file mode 100644 index 00000000..f2c46ac6 --- /dev/null +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/InternalMessageServiceImpl.java @@ -0,0 +1,123 @@ +package com.alops.system.service.impl; + +import java.util.Collections; +import java.util.List; + +import com.alops.common.utils.DateUtils; +import com.alops.system.domain.InternalMessage; +import com.alops.system.mapper.InternalMessageMapper; +import com.alops.system.service.IInternalMessageService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + + +/** + * 站内信Service业务层处理 + * + * @author ruoyi + * @date 2025-10-23 + */ +@Service +public class InternalMessageServiceImpl implements IInternalMessageService +{ + @Autowired + private InternalMessageMapper internalMessageMapper; + + @Override + public List getMessagesByUserId(Long userId) { + if (userId == null) { + return Collections.emptyList(); + } + return internalMessageMapper.selectMessagesByUserId(userId); + } + @Override + public int getUnreadCountByUserId(Long userId) { + if (userId == null) { + return 0; + } + return internalMessageMapper.countUnreadByUserId(userId); + } + /** + * 查询站内信 + * + * @param id 站内信主键 + * @return 站内信 + */ + @Override + public InternalMessage selectInternalMessageById(Long id) + { + return internalMessageMapper.selectInternalMessageById(id); + } + + @Transactional(rollbackFor = Exception.class) + public int markMessagesAsRead(Long[] ids) { + if (ids == null || ids.length == 0) { + return 0; + } + int updatedRows = internalMessageMapper.updateMessageStatusToRead(ids); + return updatedRows; + } + + + /** + * 查询站内信列表 + * + * @param internalMessage 站内信 + * @return 站内信 + */ + @Override + public List selectInternalMessageList(InternalMessage internalMessage) + { + return internalMessageMapper.selectInternalMessageList(internalMessage); + } + + /** + * 新增站内信 + * + * @param internalMessage 站内信 + * @return 结果 + */ + @Override + public int insertInternalMessage(InternalMessage internalMessage) + { + internalMessage.setCreateTime(DateUtils.getNowDate()); + return internalMessageMapper.insertInternalMessage(internalMessage); + } + + /** + * 修改站内信 + * + * @param internalMessage 站内信 + * @return 结果 + */ + @Override + public int updateInternalMessage(InternalMessage internalMessage) + { + return internalMessageMapper.updateInternalMessage(internalMessage); + } + + /** + * 批量删除站内信 + * + * @param ids 需要删除的站内信主键 + * @return 结果 + */ + @Override + public int deleteInternalMessageByIds(Long[] ids) + { + return internalMessageMapper.deleteInternalMessageByIds(ids); + } + + /** + * 删除站内信信息 + * + * @param id 站内信主键 + * @return 结果 + */ + @Override + public int deleteInternalMessageById(Long id) + { + return internalMessageMapper.deleteInternalMessageById(id); + } +} diff --git a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java index e4b85269..ecf3d3b0 100644 --- a/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java +++ b/ALOps_sys_backend/alops-system/src/main/java/com/alops/system/service/impl/ScheduleRulesServiceImpl.java @@ -150,16 +150,16 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { newRule.setStatus(2); // 执行成功 scheduleRulesMapper.updateScheduleRules(newRule); - String content = String.format("立即任务 [%s] 执行完成", newRule.getRuleName()); - iAlertSenderService.sendInternalMessage(content, "2"); + String content = String.format("立即任务执行完成", newRule.getRuleName()); + iAlertSenderService.sendInternalMessage("采集任务成功",content, "2"); } catch (Exception e) { newRule.setStatus(3); // 执行失败 scheduleRulesMapper.updateScheduleRules(newRule); - String content = String.format("立即任务 [%s] 执行失败：%s", newRule.getRuleName(), e.getMessage()); + String content = String.format("立即任务执行失败：%s", e.getMessage()); try { - iAlertSenderService.sendInternalMessage(content, "2"); + iAlertSenderService.sendInternalMessage("采集任务失败",content, "2"); } catch (Exception ne) { log.error("发送站内信失败: {}", ne.getMessage(), ne); } @@ -203,35 +203,128 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { equBaseMapper.deleteEquVlan(); } //往上抛异常 +// @Override +// public int executeImmediately() throws ExecutionException, InterruptedException { +// // 删除之前的信息 +// deleteAllEquipmentData(); +// +// List devices = equBaseMapper.findEquBaseList(); +// if (devices.isEmpty()) return Code.FAIL; +// +// Map> oidMap = new HashMap<>(); +// for (EquBase dev : devices) { +// oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId())); +// } +// +// ExecutorService executor = Executors.newFixedThreadPool(10); +// List> futures = new ArrayList<>(); +// +// for (EquBase device : devices) { +// futures.add(executor.submit(() -> { +// long start = System.currentTimeMillis(); +// +// //获取对应oid +// List oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList()); +// if (oids.isEmpty()) { +// System.err.println("⚠️ 设备[" + device.getName() + "]未配置OID，跳过"); +// return null; +// } +// //采集对应结果 +// Map snmpResults = snmpService.query(device.getIp(), oids); +// if (snmpResults == null || snmpResults.isEmpty()) { +// System.err.println("⚠️ 设备[" + device.getName() + "] SNMP采集为空，跳过"); +// return null; +// } +// +// long afterSnmp = System.currentTimeMillis(); +// System.out.println("SNMP查询耗时: " + (afterSnmp - start) + " ms"); +// //生成对应实体 +// GatherEntities entities = memoryToolService.buildAndInsertEntity(device, snmpResults); +// +// if (entities == null || entities.getEquGather() == null) { +// System.err.println("⚠️ 设备[" + device.getName() + "] 构建实体失败，跳过"); +// return null; +// } +// +// long afterBuild = System.currentTimeMillis(); +// System.out.println("buildAndInsertEntity耗时: " + (afterBuild - afterSnmp) + " ms"); +// +// //构建健康度 +// if (entities != null && entities.getEquGather() != null) { +// Double temperature = parseDouble(snmpResults.get("temperature")); +// Double cpuUsage = parseDouble(snmpResults.get("cpu_utilization")); +// Double memUsage = parseDouble(snmpResults.get("memory_utilization")); +// double tempHealth = calculateTemperatureHealth(temperature); +// double cpuHealth = calculateCpuHealth(cpuUsage); +// double memHealth = calculateMemoryHealth(memUsage); +// double totalHealth = Math.max(Math.max(tempHealth, 1), Math.max(cpuHealth, memHealth)); +// entities.getEquGather().setHealth(BigDecimal.valueOf(totalHealth)); +// return entities; +// } +// return null; +// })); +// } +// //收集采集结果，批量存入数据库 +// List allEntities = new ArrayList<>(); +// try { +// for (Future f : futures) { +// GatherEntities g = f.get(); // 阻塞等待线程完成，如果线程异常会抛ExecutionException +// if (g == null) throw new RuntimeException("设备采集返回 null，可能采集失败"); +// allEntities.add(g); +// } +// } catch (ExecutionException | InterruptedException e) { +// e.printStackTrace(); +// executor.shutdownNow(); // 出异常直接中断剩余线程 +// throw e; // 抛给调用方 Job 捕获 +// } finally { +// executor.shutdown(); +// } +// return persistAndUpdate(allEntities); +// +// } + + @Transactional(rollbackFor = Exception.class) @Override public int executeImmediately() throws ExecutionException, InterruptedException { - // 删除之前的信息 - deleteAllEquipmentData(); - + // 获取所有设备 List devices = equBaseMapper.findEquBaseList(); - if (devices.isEmpty()) return Code.FAIL; + if (devices.isEmpty()) { + log.warn("设备列表为空，任务结束"); + return Code.FAIL; + } + // 构建设备 OID 映射 Map> oidMap = new HashMap<>(); for (EquBase dev : devices) { oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId())); } + // 并发采集 ExecutorService executor = Executors.newFixedThreadPool(10); List> futures = new ArrayList<>(); - for (EquBase device : devices) { futures.add(executor.submit(() -> { - long start = System.currentTimeMillis(); - List oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList()); - Map snmpResults = snmpService.query(device.getIp(), oids); - long afterSnmp = System.currentTimeMillis(); - System.out.println("SNMP查询耗时: " + (afterSnmp - start) + " ms"); - - GatherEntities entities = memoryToolService.buildAndInsertEntity(device, snmpResults); - long afterBuild = System.currentTimeMillis(); - System.out.println("buildAndInsertEntity耗时: " + (afterBuild - afterSnmp) + " ms"); - - if (entities != null && entities.getEquGather() != null) { + log.info("开始采集设备 [{}] ({})", device.getName(), device.getIp()); + try { + List oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList()); + if (oids.isEmpty()) { + log.warn("⚠️ 设备 [{}] 未配置 OID，跳过采集", device.getName()); + return null; + } + + Map snmpResults = snmpService.query(device.getIp(), oids); + if (snmpResults == null || snmpResults.isEmpty()) { + log.warn("⚠️ 设备 [{}] SNMP 采集为空", device.getName()); + return null; + } + + GatherEntities entities = memoryToolService.buildAndInsertEntity(device, snmpResults); + if (entities == null || entities.getEquGather() == null) { + log.warn("⚠️ 设备 [{}] 构建实体失败", device.getName()); + return null; + } + + // 计算健康度示例 Double temperature = parseDouble(snmpResults.get("temperature")); Double cpuUsage = parseDouble(snmpResults.get("cpu_utilization")); Double memUsage = parseDouble(snmpResults.get("memory_utilization")); @@ -240,28 +333,59 @@ public class ScheduleRulesServiceImpl implements IScheduleRulesService { double memHealth = calculateMemoryHealth(memUsage); double totalHealth = Math.max(Math.max(tempHealth, 1), Math.max(cpuHealth, memHealth)); entities.getEquGather().setHealth(BigDecimal.valueOf(totalHealth)); + + log.info("✅ 设备 [{}] 采集请求成功", device.getName()); return entities; + } catch (Exception e) { + log.error("采集设备请求 [{}] 出错: {}", device.getName(), e.getMessage(), e); + return null; } - return null; })); } - List allEntities = new ArrayList<>(); - try { - for (Future f : futures) { - GatherEntities g = f.get(); // 阻塞等待线程完成，如果线程异常会抛ExecutionException - if (g == null) throw new RuntimeException("设备采集返回 null，可能采集失败"); - allEntities.add(g); + // 收集采集结果 + List successList = new ArrayList<>(); + List failed = new ArrayList<>(); + for (int i = 0; i < futures.size(); i++) { + EquBase dev = devices.get(i); + try { + GatherEntities g = futures.get(i).get(); + if (g != null) successList.add(g); + else failed.add(dev.getName()); + } catch (Exception e) { + failed.add(dev.getName()); + log.error("收集设备 [{}] 结果失败", dev.getName(), e); } - } catch (ExecutionException | InterruptedException e) { - e.printStackTrace(); - executor.shutdownNow(); // 出异常直接中断剩余线程 - throw e; // 抛给调用方 Job 捕获 - } finally { - executor.shutdown(); } - return persistAndUpdate(allEntities); + executor.shutdown(); + + // ✅ 完整性校验：找出所有子表为空的设备 + List invalidDevices = successList.stream() + .filter(g -> g.getSubEntities() == null + || g.getSubEntities().isEmpty() + || g.getSubEntities().values().stream().anyMatch(list -> list == null || list.isEmpty())) + .map(g -> g.getEquGather().getEquId()) + .collect(Collectors.toList()); + + // 如果全部失败或者发现不完整设备，抛异常 + if (successList.isEmpty() || !invalidDevices.isEmpty()) { + String msg = "本次采集失败，保持旧数据完好。失败设备：" + + String.join(", ", failed) + + (invalidDevices.isEmpty() ? "" : "，采集数据不完整设备：" + String.join(", ", invalidDevices)); + log.error(msg); + throw new RuntimeException(msg); // 异常抛出去，外层可捕获发送邮件 + } + // 事务内：删除旧表 + 插入成功设备数据 + deleteAllEquipmentData(); // 删除依附表，不含 equ_gather + int inserted = persistAndUpdate(successList); // 插入成功设备数据 + + if (inserted <= 0) { + throw new RuntimeException("成功采集设备写入失败，回滚事务"); + } + //发送一个采集成功的信 + log.info("本次采集成功 {} 台，删除旧表并写入最新数据，失败设备：{}", successList.size(), failed); + return Code.SUCCESS; } /** diff --git a/ALOps_sys_backend/alops-system/src/main/resources/mapper/system/InternalMessageMapper.xml b/ALOps_sys_backend/alops-system/src/main/resources/mapper/system/InternalMessageMapper.xml index e1890bff..0c4a20f6 100644 --- a/ALOps_sys_backend/alops-system/src/main/resources/mapper/system/InternalMessageMapper.xml +++ b/ALOps_sys_backend/alops-system/src/main/resources/mapper/system/InternalMessageMapper.xml @@ -10,22 +10,51 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" + - - INSERT INTO internal_message (user_id, content, status, create_time) + + + + + + + + UPDATE internal_message + SET status = 'read' + WHERE id IN + + #{id} + + AND status = 'unread' + + + + + + INSERT INTO internal_message (user_id, content, status, create_time,title) VALUES - (#{msg.userId}, #{msg.content}, #{msg.status}, #{msg.createTime}) + (#{msg.userId}, #{msg.content}, #{msg.status}, #{msg.createTime}, #{msg.title}) - select id, user_id, content, status, create_time from internal_message + select id, user_id, content, status, create_time,title from internal_message