#新版告警修改

hzims-service-api/alarm-api/src/main/java/com/hnac/hzims/business/ws/alart/vo/AlartParamVo.java

@@ -10,12 +10,9 @@ import java.util.List;
  * @author ysj
  */
 @Data
-@ApiModel(value = "告警弹框参数", description = "")
+@ApiModel(value = "告警websocket参数", description = "")
 public class AlartParamVo {
 
-	@ApiModelProperty("站点类型: 2-云服务 1-代运维")
+	@ApiModelProperty("站点集合")
-	private Integer serveType;
+	private List<String> codes;
-
-	@ApiModelProperty("用户权限机构集合")
-	private List<Long> depts;
 }

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/monitor/listener/AlarmListener.java

@@ -12,6 +12,9 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Service;
 
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ThreadPoolExecutor;
+
 /**
  * @author ysj
  */
@@ -21,6 +24,9 @@ import org.springframework.stereotype.Service;
 public class AlarmListener implements IQueueConsume {
 
     @Autowired
+    private ThreadPoolExecutor pool;
+
+    @Autowired
     private MessageService messageService;
 
     @Autowired
@@ -36,22 +42,30 @@ public class AlarmListener implements IQueueConsume {
 
         // 步骤2.websocket消息推送
         if(ObjectUtil.isNotEmpty(alarm.getIsRightTabulation()) && alarm.getIsRightTabulation() == 0){
-
+            CompletableFuture.runAsync(() -> {
+                messageService.webRightMessage(alarm);
+            },pool);
         }
 
         // 步骤3.WEB/APP消息推送
         if(ObjectUtil.isNotEmpty(alarm.getIsPlatformMessage()) && alarm.getIsPlatformMessage() == 0){
+            CompletableFuture.runAsync(() -> {
                 messageService.webAppMessage(alarm);
+            },pool);
         }
 
         // 步骤4.短信推送
         if (ObjectUtil.isNotEmpty(alarm.getIsShortMessage()) && alarm.getIsShortMessage() == 0) {
+            CompletableFuture.runAsync(() -> {
                  messageService.shortMessage(alarm);
+            },pool);
         }
 
         // 步骤5.微信公众号发送
         if (ObjectUtil.isNotEmpty(alarm.getIsWxMessage()) && alarm.getIsWxMessage() == 0) {
+            CompletableFuture.runAsync(() -> {
                 messageService.weChatMessage(alarm);
+            },pool);
         }
     }
 }

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/monitor/listener/AlarmPoolManager.java

@@ -0,0 +1,34 @@
+
+package com.hnac.hzims.alarm.monitor.listener;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.*;
+
+/**
+ * @author: ysj
+ */
+@Component
+public class AlarmPoolManager {
+
+    @Bean
+    public ThreadPoolExecutor threadPoolExecutor() {
+        // 核心线程数
+        int corePoolSize = 8;
+        // 最大线程数
+        int maximumPoolSize = 16;
+        // 线程空闲时的存活时间
+        long keepAliveTime = 60L;
+        // 时间单位
+        TimeUnit unit = TimeUnit.SECONDS;
+        // 任务队列
+        BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>(100);
+        // 线程工厂
+        ThreadFactory threadFactory = Executors.defaultThreadFactory();
+        // 等待策略
+        RejectedExecutionHandler handler = new ThreadPoolExecutor.CallerRunsPolicy();
+        return new ThreadPoolExecutor(corePoolSize,maximumPoolSize,keepAliveTime,unit,workQueue,threadFactory,handler);
+    }
+
+}

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/show/service/AlarmService.java

@@ -22,7 +22,7 @@ public interface AlarmService extends IService<AlarmEntity> {
 
     List<AlarmEntity> broadcast(String startTime, String endTime,Integer serveType);
 
-    TextMessage majorAlarm(AlartParamVo param);
+    TextMessage majorAlarm(List<String> stations);
 
     List<AlarmVideoVo> alarmVideos(String stationCode, String deviceCode, String realId,Integer alarmType);
 }

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/show/service/impl/AlarmServiceImpl.java

@@ -36,6 +36,7 @@ import org.apache.commons.collections4.MapUtils;
 import org.springblade.core.log.exception.ServiceException;
 import org.springblade.core.tool.api.R;
 import org.springblade.core.tool.utils.CollectionUtil;
+import org.springblade.core.tool.utils.DateUtil;
 import org.springblade.core.tool.utils.ObjectUtil;
 import org.springblade.core.tool.utils.StringUtil;
 import org.springblade.message.fegin.IMessageClient;
@@ -138,6 +139,18 @@ public class AlarmServiceImpl extends ServiceImpl<AlarmMapper, AlarmEntity> impl
     }
 
     /**
+     * 获取权限站点
+     * @return
+     */
+    private List<StationEntity> stationsByCodes(List<String> stations) {
+        R<List<StationEntity>> result = stationClient.querySatationByCodes(stations);
+        if(!result.isSuccess() || CollectionUtil.isEmpty(result.getData())){
+            return new ArrayList<>();
+        }
+        return result.getData();
+    }
+
+    /**
      * 查询告警数量
      * @return
      */
@@ -219,26 +232,25 @@ public class AlarmServiceImpl extends ServiceImpl<AlarmMapper, AlarmEntity> impl
 
     /**
      * 弹框告警
-     * @param param
+     * @param codes
      * @return
      */
     @Override
-    public TextMessage majorAlarm(AlartParamVo param) {
+    public TextMessage majorAlarm(List<String> codes) {
         // 站点查询
-        List<StationEntity> stations = this.stations(param.getServeType());
+        List<StationEntity> stations = this.stationsByCodes(codes);
-        if(CollectionUtil.isEmpty(stations) || CollectionUtil.isEmpty(param.getDepts())){
+        if(CollectionUtil.isEmpty(stations)){
-            return new TextMessage(new ArrayList().toString());
-        }
-        List<String> effectives = stations.stream().filter(station->param.getDepts().contains(station.getRefDept())).map(StationEntity::getCode).collect(Collectors.toList());
-        if(CollectionUtil.isEmpty(effectives)){
             return new TextMessage(new ArrayList().toString());
         }
         // 条件过滤
         QueryWrapper<AlarmEntity> queryWrapper = new QueryWrapper<>();
-        queryWrapper.lambda().in(AlarmEntity::getStationId,effectives);
+        queryWrapper.lambda().in(AlarmEntity::getStationId,stations.stream().map(StationEntity::getCode).collect(Collectors.toList()));
-        queryWrapper.lambda().in(AlarmEntity::getAlarmType, Arrays.asList(AlarmConstants.FAULT,AlarmConstants.EARLY));
+        queryWrapper.lambda().in(AlarmEntity::getAlarmType, Arrays.asList(AlarmConstants.WARNING,AlarmConstants.FAULT,AlarmConstants.OFFSIDE,AlarmConstants.INTERRUPT,AlarmConstants.ABNORMAL,AlarmConstants.EARLY,AlarmConstants.START,AlarmConstants.STOP));
-        queryWrapper.lambda().eq(AlarmEntity::getIsShowAlert,0);
+        queryWrapper.lambda().eq(AlarmEntity::getIsRightTabulation,0);
-        queryWrapper.lambda().eq(AlarmEntity::getStatus,0);
+        queryWrapper.lambda().ge(AlarmEntity::getAlarmTime, DateUtil.format(new Date(),DateUtil.PATTERN_DATE) + " 00:00:00");
+        queryWrapper.lambda().le(AlarmEntity::getAlarmTime,DateUtil.format(new Date(),DateUtil.PATTERN_DATETIME));
+        queryWrapper.lambda().orderByDesc(AlarmEntity::getAlarmTime);
+        //queryWrapper.lambda().eq(AlarmEntity::getStatus,0);
         // 处理告警过滤 ：当天处理告警
         List<String> handles = alarmHandleService.handles();
         if(!CollectionUtil.isEmpty(handles)){
@@ -248,7 +260,7 @@ public class AlarmServiceImpl extends ServiceImpl<AlarmMapper, AlarmEntity> impl
         if(CollectionUtil.isEmpty(alarms)){
             return new TextMessage(new ArrayList().toString());
         }
-        return new TextMessage(JSONObject.toJSONString(alarms.stream().sorted(Comparator.comparing(AlarmEntity::getAlarmTime).reversed()).collect(Collectors.toList())));
+        return new TextMessage(JSONObject.toJSONString(alarms.stream().skip(0).limit(1000).collect(Collectors.toList())));
     }
 
     /**

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/source/service/MessageService.java

@@ -7,6 +7,8 @@ import com.hnac.hzims.alarm.config.entity.AlarmEntity;
  */
 public interface MessageService {
 
+    void webRightMessage(AlarmEntity alarm);
+
     void webAppMessage(AlarmEntity entity);
 
     void shortMessage(AlarmEntity entity);

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/source/service/impl/MessageServiceImpl.java

@@ -1,8 +1,10 @@
 package com.hnac.hzims.alarm.source.service.impl;
 
+import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.toolkit.ObjectUtils;
 import com.hnac.hzims.alarm.config.entity.AlarmEntity;
 import com.hnac.hzims.alarm.source.service.MessageService;
+import com.hnac.hzims.alarm.ws.alart.AlarmSessionManager;
 import com.hnac.hzims.common.constant.CommonConstant;
 import com.hnac.hzims.operational.station.entity.StationEntity;
 import com.hnac.hzims.operational.station.feign.IStationClient;
@@ -17,13 +19,16 @@ import org.springblade.message.dto.SmsImmediatelyPushDTO;
 import org.springblade.message.dto.WxMessageDTO;
 import org.springblade.message.fegin.IMessageClient;
 import org.springblade.system.cache.DictCache;
+import org.springblade.system.feign.IDeptClient;
 import org.springblade.system.feign.ISysClient;
 import org.springblade.system.user.entity.User;
 import org.springblade.system.user.entity.UserInfo;
 import org.springblade.system.user.feign.IUserClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
+import org.springframework.web.socket.TextMessage;
 
+import java.io.IOException;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.stream.Collectors;
@@ -44,6 +49,8 @@ public class MessageServiceImpl implements MessageService {
 
     private final IUserClient userClient;
 
+    private final IDeptClient deptClient;
+
     private final IMessageClient messageClient;
 
     private final IStationClient stationClient;
@@ -56,6 +63,34 @@ public class MessageServiceImpl implements MessageService {
     private String wxPushTemplate;
     @Value("${hzims.wxPush.enabled}")
     private Boolean wxPushEnabled;
+
+    /**
+     * web告警展示消息推送
+     * @param alarm
+     */
+    @Override
+    public void webRightMessage(AlarmEntity alarm) {
+        if(CollectionUtil.isEmpty(AlarmSessionManager.SESSION_POOL)){
+            return;
+        }
+        // 遍历websocket链接,拥有告警站点权限进行推送消息
+        AlarmSessionManager.SESSION_POOL.forEach((key, session) -> {
+            if(!session.getAttributes().containsKey("codes")){
+                return;
+            }
+            List<String> codes = (List<String>) session.getAttributes().get("codes");
+            if(CollectionUtil.isEmpty(codes) || !codes.contains(alarm.getStationId())){
+                return;
+            }
+            try {
+                session.sendMessage(new TextMessage(JSONObject.toJSONString(Collections.singletonList(alarm))));
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
+        });
+    }
+
+
     /**
      * web/app消息推送
      * @param entity : 告警对象
@@ -161,8 +196,6 @@ public class MessageServiceImpl implements MessageService {
                 return;
             }
             userList.addAll(result.getData());
-            log.info("短信发送用户列表：{}",userList);
-            log.info("短信发送用户列表,{}",userList);
         }else {
             //内测只推给对应的3个用户
             List<String> phoneList = Arrays.asList("18351807087", "18163793336", "18285121497");
@@ -172,9 +205,9 @@ public class MessageServiceImpl implements MessageService {
                     userList.add(userByPhone.getData().getUser());
                 }
             }
+        }
         log.info("短信发送用户列表：{}",userList);
         log.info("短信发送用户列表,{}",userList);
-        }
         String userIds = userList.stream().map(o -> String.valueOf(o.getId())).distinct().collect(Collectors.joining(","));
         WxMessageDTO message = new WxMessageDTO();
         // 模板Id
@@ -191,8 +224,8 @@ public class MessageServiceImpl implements MessageService {
         message.setTaskId(entity.getId());
         // 微信参数键值对
         HashMap<String,Object> map = new HashMap<>();
-        map.put("thing18",getTruncateString(entity.getStationName(), 20));
+        map.put("thing18",getTruncateString(entity.getStationName()));
-        map.put("thing11",getTruncateString(entity.getAlarmContext(), 20));
+        map.put("thing11",getTruncateString(entity.getAlarmContext()));
         map.put("time2",entity.getAlarmTime());
         map.put("thing14", DictCache.getValue("alarm_type", String.valueOf(entity.getAlarmType())));
         DateTimeFormatter format = DateTimeFormatter.ofPattern(DateUtil.PATTERN_DATETIME);
@@ -211,11 +244,12 @@ public class MessageServiceImpl implements MessageService {
         message.setTenantId(station.getData().getTenantId());
         messageClient.sendWxMessage(message);
     }
-    private String getTruncateString(String ruleDefName, int maxSize) {
+
-        if (ruleDefName.length()<= maxSize){
+    private String getTruncateString(String ruleDefName) {
+        if (ruleDefName.length()<= 20){
             return ruleDefName;
         }else {
-            return ruleDefName.substring(0, maxSize - 3) + "...";
+            return ruleDefName.substring(0, 20 - 3) + "...";
         }
     }
 }

hzims-service/hzims-alarm/src/main/java/com/hnac/hzims/alarm/ws/alart/AlarmHandler.java

@@ -4,11 +4,9 @@ import com.alibaba.fastjson.JSONObject;
 import com.hnac.hzims.alarm.show.service.AlarmService;
 import com.hnac.hzims.business.ws.alart.vo.AlartParamVo;
 import lombok.extern.slf4j.Slf4j;
+import org.springblade.core.tool.utils.CollectionUtil;
 import org.springblade.core.tool.utils.ObjectUtil;
 import org.springblade.core.tool.utils.StringUtil;
-import org.springblade.system.feign.IDeptClient;
-import org.springblade.system.feign.ISysClient;
-import org.springblade.system.user.feign.IUserClient;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.socket.CloseStatus;
 import org.springframework.web.socket.TextMessage;
@@ -16,6 +14,8 @@ import org.springframework.web.socket.WebSocketSession;
 import org.springframework.web.socket.handler.TextWebSocketHandler;
 
 import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
 
 /**
  * @author ysj
@@ -56,11 +56,13 @@ public class AlarmHandler extends TextWebSocketHandler {
 		if(StringUtil.isBlank(message)){
 			return;
 		}
-		AlartParamVo param = JSONObject.parseObject(message,AlartParamVo.class);
+		AlartParamVo param = JSONObject.parseObject(textMessage.getPayload(), AlartParamVo.class);
-		if(ObjectUtil.isEmpty(param)){
+		if(ObjectUtil.isEmpty(param) || CollectionUtil.isEmpty(param.getCodes())){
 			return;
 		}
-		TextMessage sendMessage = alarmService.majorAlarm(param);
+		session.getAttributes().put("codes", param.getCodes());
+		AlarmSessionManager.SESSION_POOL.put(userId,session);
+		TextMessage sendMessage = alarmService.majorAlarm(param.getCodes());
 		session.sendMessage(sendMessage);
 	}
 }