Merge branch 'dev' of https://gitea.wangzhiwen.top/0214/ALOps_sys1 into dev

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<InternalMessage> messages = iInternalMessageService.getMessagesByUserId(userId);
+
+        return AjaxResult.success(messages);
+    }
+
+
+
+
+
+}

ALOps_sys_backend/alops-framework/pom.xml

@@ -60,6 +60,7 @@
             <groupId>com.alops</groupId>
             <artifactId>alops-system</artifactId>
         </dependency>
+
         <dependency>
             <groupId>org.snmp4j</groupId>
             <artifactId>snmp4j</artifactId>

ALOps_sys_backend/alops-system/pom.xml

@@ -60,6 +60,19 @@
                 </exclusion>
             </exclusions>
         </dependency>
+
+        <!-- websocket -->
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+
+
         <dependency>
             <groupId>org.quartz-scheduler</groupId>
             <artifactId>quartz</artifactId>
@@ -79,6 +92,10 @@
             <artifactId>lombok</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-websocket</artifactId>
+        </dependency>
 
 
     </dependencies>

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("*");
+
+    }
+}

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();
     }
 }

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<Long, WebSocketSession> 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<String, String> 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<String, Object> 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<String, Object> 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<Long> getOnlineUserIds() {
+        return onlineUsers.keySet();
+    }
+
+
+
+}

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);
+//                scheduler.deleteJob(jobKey);
-                String Content=String.format("任务执行完此次，已取消后续任务");
+                boolean deleted = scheduler.deleteJob(jobKey);
-
+                if (deleted) {
-                iAlertSenderService.sendInternalMessage( Content,"2" );
+                    iAlertSenderService.sendInternalMessage("采集任务取消成功","任务执行完此次，已取消后续任务","2");
-                log.info("规则[{}] 已标记为取消状态(4)，对应定时任务已删除: {}", rule.getId(), jobKey);
+                    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) {

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<InternalMessage> selectMessagesByUserId(@Param("userId") Long userId);
+            /**
+             * 统计某个用户的未读消息数量
+             */
+            int countUnreadByUserId(@Param("userId") Long userId);
+
+            /**
+             * 批量更新消息状态为已读
+             * @param ids 消息ID数组
+             * @return 更新行数
+             */
+            int updateMessageStatusToRead( Long[] ids);
             /**
          * 批量插入消息
          * @param messages 消息列表

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<String> alertMessage, String alertWay, String target);
-    public void sendInternalMessage(String content, String roleIdStr);
+    public void sendInternalMessage(String title, String content, String roleIdStr);
 }

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<InternalMessage> 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<InternalMessage> 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);
+}

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<AlertMessage> messages) {
+    public void insertAlertMessages(List<AlertMessage> 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) {

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.*;
-import java.util.Date;
-import java.util.List;
 
 @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
+        //运维人员2
-        sendInternalMessage(combinedMessage,"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<SysUser> users = userMapper.selectByRoleId(roleId);
+//        if (users == null || users.isEmpty()) return;
+//        //1.插入站内信表
+//        List<InternalMessage> 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<SysUser> users = userMapper.selectByRoleId(roleId);
         if (users == null || users.isEmpty()) return;
-
+        //1.插入站内信表
         List<InternalMessage> 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<Map<String, Object>> result =  insertMessages(messages);
+
+        //2.发送实时消息
+        for (Map<String, Object> item : result) {
+            Long userId = (Long) item.get("userId");
+            Long messageId = (Long) item.get("id");
 
-        if (!messages.isEmpty()) {
+            // 调用 WebSocket 工具发送
-            insertMessages(messages);
+            WebSocketServer.sendAlertMessage(userId, messageId, title, content);
         }
+
+        log.info("共发送站内信 {} 条（角色ID={}）", result.size(), roleId);
     }
 
+
     @Transactional(rollbackFor = Exception.class)
-    public void insertMessages(List<InternalMessage> messages) {
+    public List<Map<String, Object>> insertMessages(List<InternalMessage> 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<Map<String, Object>> resultList = new ArrayList<>();
+        for (InternalMessage msg : messages) {
+            Map<String, Object> map = new HashMap<>();
+            map.put("id", msg.getId());         // 主键已被MyBatis自动回填
+            map.put("userId", msg.getUserId());
+            resultList.add(map);
+        }
+
+        return resultList;
     }
 }
 
+

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<InternalMessage> 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<InternalMessage> 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);
+    }
+}

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());
+                            String content = String.format("立即任务执行完成", newRule.getRuleName());
-                            iAlertSenderService.sendInternalMessage(content, "2");
+                            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<EquBase> devices = equBaseMapper.findEquBaseList();
+//        if (devices.isEmpty()) return Code.FAIL;
+//
+//        Map<String, List<EquOid>> oidMap = new HashMap<>();
+//        for (EquBase dev : devices) {
+//            oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId()));
+//        }
+//
+//        ExecutorService executor = Executors.newFixedThreadPool(10);
+//        List<Future<GatherEntities>> futures = new ArrayList<>();
+//
+//        for (EquBase device : devices) {
+//            futures.add(executor.submit(() -> {
+//                long start = System.currentTimeMillis();
+//
+//              //获取对应oid
+//                List<EquOid> oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList());
+//                if (oids.isEmpty()) {
+//                    System.err.println("⚠️ 设备[" + device.getName() + "]未配置OID，跳过");
+//                    return null;
+//                }
+//               //采集对应结果
+//                Map<String, Object> 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<GatherEntities> allEntities = new ArrayList<>();
+//        try {
+//            for (Future<GatherEntities> 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<EquBase> devices = equBaseMapper.findEquBaseList();
-        if (devices.isEmpty()) return Code.FAIL;
+        if (devices.isEmpty()) {
+            log.warn("设备列表为空，任务结束");
+            return Code.FAIL;
+        }
 
+        // 构建设备 OID 映射
         Map<String, List<EquOid>> oidMap = new HashMap<>();
         for (EquBase dev : devices) {
             oidMap.put(dev.getEquManuId(), equOidMapper.findByManuIdAndStatus(dev.getEquManuId()));
         }
 
+        // 并发采集
         ExecutorService executor = Executors.newFixedThreadPool(10);
         List<Future<GatherEntities>> futures = new ArrayList<>();
-
         for (EquBase device : devices) {
             futures.add(executor.submit(() -> {
-                long start = System.currentTimeMillis();
+                log.info("开始采集设备 [{}] ({})", device.getName(), device.getIp());
-                List<EquOid> oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList());
+                try {
-                Map<String, Object> snmpResults = snmpService.query(device.getIp(), oids);
+                    List<EquOid> oids = oidMap.getOrDefault(device.getEquManuId(), Collections.emptyList());
-                long afterSnmp = System.currentTimeMillis();
+                    if (oids.isEmpty()) {
-                System.out.println("SNMP查询耗时: " + (afterSnmp - start) + " ms");
+                        log.warn("⚠️ 设备 [{}] 未配置 OID，跳过采集", device.getName());
-
+                        return null;
-                GatherEntities entities = memoryToolService.buildAndInsertEntity(device, snmpResults);
+                    }
-                long afterBuild = System.currentTimeMillis();
+
-                System.out.println("buildAndInsertEntity耗时: " + (afterBuild - afterSnmp) + " ms");
+                    Map<String, Object> snmpResults = snmpService.query(device.getIp(), oids);
-
+                    if (snmpResults == null || snmpResults.isEmpty()) {
-                if (entities != null && entities.getEquGather() != null) {
+                        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<GatherEntities> allEntities = new ArrayList<>();
+        // 收集采集结果
-        try {
+        List<GatherEntities> successList = new ArrayList<>();
-            for (Future<GatherEntities> f : futures) {
+        List<String> failed = new ArrayList<>();
-                GatherEntities g = f.get(); // 阻塞等待线程完成，如果线程异常会抛ExecutionException
+        for (int i = 0; i < futures.size(); i++) {
-                if (g == null) throw new RuntimeException("设备采集返回 null，可能采集失败");
+            EquBase dev = devices.get(i);
-                allEntities.add(g);
+            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<String> 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;
     }
 
     /**

ALOps_sys_backend/alops-system/src/main/resources/mapper/system/InternalMessageMapper.xml

@@ -10,22 +10,51 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
         <result property="content"    column="content"    />
         <result property="status"    column="status"    />
         <result property="createTime"    column="create_time"    />
+        <result property="title"    column="title"    />
 
     </resultMap>
 
 
-    <insert id="batchInsert" parameterType="java.util.List">
+    <select id="selectMessagesByUserId" resultMap="InternalMessageResult">
-        INSERT INTO internal_message (user_id, content, status, create_time)
+        SELECT *
+        FROM internal_message
+        WHERE user_id = #{userId}
+        ORDER BY create_time DESC
+    </select>
+
+    <select id="countUnreadByUserId" resultType="int">
+        SELECT COUNT(*)
+        FROM internal_message
+        WHERE user_id = #{userId}
+          AND status = 'unread'
+    </select>
+
+
+    <!-- 批量更新消息状态为已读 -->
+    <update id="updateMessageStatusToRead" parameterType="long[]">
+        UPDATE internal_message
+        SET status = 'read'
+        WHERE id IN
+        <foreach collection="array" item="id" open="(" separator="," close=")">
+            #{id}
+        </foreach>
+        AND status = 'unread'
+    </update>
+
+
+
+    <insert id="batchInsert" parameterType="java.util.List" useGeneratedKeys="true" keyProperty="id">
+        INSERT INTO internal_message (user_id, content, status, create_time,title)
         VALUES
         <foreach collection="list" item="msg" separator=",">
-            (#{msg.userId}, #{msg.content}, #{msg.status}, #{msg.createTime})
+            (#{msg.userId}, #{msg.content}, #{msg.status}, #{msg.createTime}, #{msg.title})
         </foreach>
     </insert>
 
 
 
     <sql id="selectInternalMessageVo">
-        select id, user_id, content, status, create_time from internal_message
+        select id, user_id, content, status, create_time,title from internal_message
     </sql>
 
     <select id="selectInternalMessageList" parameterType="InternalMessage" resultMap="InternalMessageResult">