1、延迟队列实现异常误报警；

mh-common/src/main/java/com/mh/common/core/domain/entity/AlarmRecords.java

@@ -88,6 +88,19 @@ public class AlarmRecords implements Serializable {
      */
     private int isSend;
 
+    /**
+     * 推送次数，大于3次不再推送
+     */
+    private int sendNum;
+
+    public int getSendNum() {
+        return sendNum;
+    }
+
+    public void setSendNum(int sendNum) {
+        this.sendNum = sendNum;
+    }
+
     public int getIsSend() {
         return isSend;
     }

mh-framework/src/main/java/com/mh/framework/rabbitmq/RabbitMqConfig.java

@@ -8,6 +8,9 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.HashMap;
+import java.util.Map;
+
 /**
  * @Author : Rainbow
  * @date : 2023/5/26
@@ -16,6 +19,12 @@ import org.springframework.context.annotation.Configuration;
 public class RabbitMqConfig {
     /**交换机*/
     public static final String EXCHANGE_NAME = "exchange_eemcs";
+
+    /**
+     * 延迟队列
+     */
+    public static final String DELAY_EXCHANGE_NAME = "device.delayed.exchange";
+
     /**主机队列*/
     public static final String QUEUE_CHILLER = "queue_chiller";
     /**主机routing-key*/
@@ -34,6 +43,12 @@ public class RabbitMqConfig {
     /**温湿度 routing-key*/
     public static final String ROUTING_KEY_TEMP = "topic.temp.eemcs.#";
 
+    /**
+     * 报警队列
+     */
+    public static final String QUEUE_ALARM = "device.alarm.queue";
+
+    public static final String ROUTING_KEY_ALARM = "topic.alarm.eemcs.#";
 
     /**durable参数表示交换机是否持久化，值为true表示持久化，值为false表示不持久化。
      * 在RabbitMQ中，持久化交换机会被存储在磁盘上以便在服务器重启后恢复，
@@ -120,4 +135,27 @@ public class RabbitMqConfig {
         return factory;
     }
 
+    // 延迟交换机（使用自定义类型x-delayed-message）
+    @Bean
+    public CustomExchange delayedExchange() {
+        Map<String, Object> args = new HashMap<>();
+        args.put("x-delayed-type", "direct");  // 底层转发类型
+        return new CustomExchange(DELAY_EXCHANGE_NAME, "x-delayed-message", true, false, args);
+    }
+
+    // 报警队列
+    @Bean
+    public Queue alarmQueue() {
+        return new Queue(QUEUE_ALARM, true);
+    }
+
+    // 绑定延迟交换机与队列
+    @Bean
+    public Binding binding(Queue alarmQueue, CustomExchange delayedExchange) {
+        return BindingBuilder.bind(alarmQueue)
+                .to(delayedExchange)
+                .with(ROUTING_KEY_ALARM)
+                .noargs();
+    }
+
 }

mh-framework/src/main/java/com/mh/framework/rabbitmq/consumer/ReceiveHandler.java

@@ -1,14 +1,14 @@
 package com.mh.framework.rabbitmq.consumer;
 
-import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.JSONObject;
+import com.mh.common.core.redis.RedisCache;
 import com.mh.common.model.request.AdvantechReceiver;
 import com.mh.common.model.request.OneTwoThreeTempData;
 import com.mh.framework.dealdata.DataProcessService;
 import com.mh.framework.rabbitmq.RabbitMqConfig;
+import com.mh.system.service.operation.IAlarmRecordsService;
 import com.rabbitmq.client.Channel;
 import lombok.extern.slf4j.Slf4j;
-import org.apache.poi.ss.formula.functions.T;
 import org.springframework.amqp.core.ExchangeTypes;
 import org.springframework.amqp.rabbit.annotation.Exchange;
 import org.springframework.amqp.rabbit.annotation.Queue;
@@ -33,6 +33,14 @@ public class ReceiveHandler {
     @Autowired
     DataProcessService dataProcessService;
 
+    @Autowired
+    IAlarmRecordsService alarmRecordsService;
+
+    @Autowired
+    private RedisCache redisTemplate;
+
+    private static final String ALARM_CANCEL_PREFIX = "alarm:cancel:";
+
     /**
      * 监听主机参数
      * queues：指定监听的队列名，可以接收单个队列，也可以接收多个队列的数组或列表。
@@ -165,4 +173,41 @@ public class ReceiveHandler {
         }
     }
 
+    /**
+     * 处理设备报警延时队列数据
+     *
+     * @param msg
+     * @param channel
+     * @param tag
+     */
+    @RabbitListener(bindings = @QueueBinding(
+            value = @Queue(value = RabbitMqConfig.QUEUE_ALARM, durable = "true"),
+            exchange = @Exchange(
+                    value = RabbitMqConfig.DELAY_EXCHANGE_NAME,
+                    ignoreDeclarationExceptions = "true",
+                    type = ExchangeTypes.TOPIC
+            ),
+            key = {RabbitMqConfig.ROUTING_KEY_ALARM}
+    ))
+    public void handleAlarm(@Payload String msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws InterruptedException, IOException {
+        try {
+            String deviceId = msg.split(":")[1];
+            String cancelKey = ALARM_CANCEL_PREFIX + deviceId;
+//            // 检查Redis中是否存在取消标记
+//            if (Boolean.TRUE.equals(redisTemplate.hasKey(cancelKey))) {
+//                redisTemplate.deleteObject(cancelKey);
+//                System.out.println("报警已取消: " + deviceId);
+//                return;
+//            }
+            // 执行生成报警数据
+            alarmRecordsService.insertOrUpdateAlarmRecord(deviceId);
+            // 正常执行，手动确认ack
+            channel.basicAck(tag, false);
+        } catch (Exception e) {
+            log.error("data:{},ddcException:{}", msg, e);
+            Thread.sleep(100);
+            channel.basicAck(tag, false);
+        }
+    }
+
 }

mh-framework/src/main/java/com/mh/framework/rabbitmq/producer/SendMsgByTopic.java

@@ -3,6 +3,8 @@ package com.mh.framework.rabbitmq.producer;
 import com.mh.common.model.request.AdvantechReceiver;
 import com.mh.framework.rabbitmq.RabbitMqConfig;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
@@ -40,4 +42,15 @@ public class SendMsgByTopic {
         rabbitTemplate.convertAndSend(RabbitMqConfig.EXCHANGE_NAME,"topic.temp.eemcs.data",data);
         return "success";
     }
+
+    /**延迟队列注入rabbitmq*/
+    public void sendDelayedAlarm(String deviceId, int delayMinutes) {
+        String message = "ALARM:" + deviceId;
+
+        MessageProperties props = new MessageProperties();
+        props.setHeader("x-delay", delayMinutes * 60 * 1000);  // 延迟毫秒
+        Message msg = new Message(message.getBytes(), props);
+
+        rabbitTemplate.send(RabbitMqConfig.DELAY_EXCHANGE_NAME, RabbitMqConfig.ROUTING_KEY_ALARM, msg);
+    }
 }

mh-quartz/src/main/java/com/mh/quartz/task/CreateAlarmTask.java

@@ -6,18 +6,21 @@ import com.mh.common.core.domain.entity.AlarmRules;
 import com.mh.common.core.domain.entity.CollectionParamsManage;
 import com.mh.common.utils.BigDecimalUtils;
 import com.mh.common.utils.DateUtils;
+import com.mh.framework.rabbitmq.producer.SendMsgByTopic;
 import com.mh.system.service.device.ICollectionParamsManageService;
 import com.mh.system.service.operation.IAlarmCodeService;
 import com.mh.system.service.operation.IAlarmRecordsService;
 import com.mh.system.service.operation.IAlarmRulesService;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 
 /**
@@ -31,84 +34,25 @@ import java.util.List;
 @Component("createAlarmTask")
 public class CreateAlarmTask {
 
-    private final IAlarmRulesService alarmRulesService;
-
-    private final IAlarmCodeService alarmCodeService;
-
-    private final ICollectionParamsManageService collectionParamsManageService;
 
     private final IAlarmRecordsService alarmRecordService;
 
+    private final SendMsgByTopic sendMsgByTopic;
+
     @Autowired
-    public CreateAlarmTask(IAlarmRulesService alarmRulesService, IAlarmCodeService alarmCodeService, ICollectionParamsManageService collectionParamsManageService, IAlarmRecordsService alarmRecordService) {
+    public CreateAlarmTask(IAlarmRecordsService alarmRecordService, SendMsgByTopic sendMsgByTopic) {
-        this.alarmRulesService = alarmRulesService;
-        this.alarmCodeService = alarmCodeService;
-        this.collectionParamsManageService = collectionParamsManageService;
         this.alarmRecordService = alarmRecordService;
+        this.sendMsgByTopic = sendMsgByTopic;
     }
 
     public void createAlarmTask() {
         log.info("创建报警记录");
-        // 查询仪表报警规则记录，查看哪些规则启用了
+        // 第一次判断
-        List<AlarmRules> alarmRules = alarmRulesService.selectAlarmRulesListByStatus(0);
+        String alarmTask = alarmRecordService.createAlarmTask();
-        // 循环查询报警规则，判断是否满足报警条件
+        if (null == alarmTask) {
-        for (AlarmRules alarmRule : alarmRules) {
+            return;
-            // 判断报警类型
-            if ("0".equals(alarmRule.getAlarmType())) {
-                // 当前是越限事件
-                // 查询事件类型查询对应的报警模板内容
-                AlarmCode alarmCode = alarmCodeService.selectAlarmCodeByAlarmType(alarmRule.getEventType());
-                // 获取当前采集参数值
-                CollectionParamsManage collectionParamsManage = collectionParamsManageService.selectCollectionParamsManageById(alarmRule.getCpmId());
-                // 判断当前值是否是当前事件
-                AlarmRecords alarmRecords = new AlarmRecords();
-                BigDecimal curValue = collectionParamsManage.getCurValue();
-                Date curTime = collectionParamsManage.getCurTime();
-                // 阈值
-                String threshold1 = alarmRule.getThreshold1();
-                if (alarmRule.getTimePeriodSet() == 0 && DateUtils.isSameDay(curTime, new Date())) {
-                    // 执行相关操作
-                    insertOrUpdateRecord(alarmRule, curValue, threshold1, alarmCode, alarmRecords, collectionParamsManage);
-                } else if (alarmRule.getTimePeriodSet() == 1
-                        && DateUtils.isSameDay(collectionParamsManage.getCurTime(), new Date())
-                        && DateUtils.isCurrentTimeInRange(alarmRule.getBeginTime(), alarmRule.getEndTime(), curTime)
-                ) {
-                    // 执行相关操作
-                    insertOrUpdateRecord(alarmRule, curValue, threshold1, alarmCode, alarmRecords, collectionParamsManage);
-                }
-            }
-        }
-    }
-
-    private void insertOrUpdateRecord(AlarmRules alarmRule, BigDecimal curValue, String threshold1, AlarmCode alarmCode, AlarmRecords alarmRecords, CollectionParamsManage collectionParamsManage) {
-        boolean compare = BigDecimalUtils.compare(alarmRule.getCondition1(), curValue, new BigDecimal(threshold1));
-        if (compare) {
-            // 创建报警记录
-            String content = alarmCode.getMsgContent();
-            content = content.replace("#{curValue}", curValue.setScale(1, RoundingMode.HALF_UP).toString());
-            content = content.replace("#{setValue}", alarmRule.getCondition1() + threshold1);
-            alarmRecords.setContent(content);
-            alarmRecords.setAlarmType(alarmRule.getAlarmType());
-            alarmRecords.setEventType(alarmRule.getEventType());
-            alarmRecords.setAlarmLevel(alarmRule.getAlarmLevel());
-            alarmRecords.setLedgerId(alarmRule.getLedgerId());
-            alarmRecords.setCpmId(alarmRule.getCpmId());
-            alarmRecords.setDeviceName(alarmRule.getDeviceName());
-            alarmRecords.setCpmName(alarmRule.getCpmName());
-            alarmRecords.setCreateTime(collectionParamsManage.getCurTime());
-            // 判断报警记录是否已经存在
-            AlarmRecords isExits = alarmRecordService.selectIsExist(alarmRecords);
-            if (isExits == null) {
-                alarmRecordService.insertAlarmRecord(alarmRecords);
-            } else {
-                // 更新报警记录
-                isExits.setContent(content);
-                isExits.setCreateTime(collectionParamsManage.getCurTime());
-                isExits.setStatus(0);
-                isExits.setIsSend(0);
-                alarmRecordService.updateAlarmRecord(isExits);
-            }
         }
+        sendMsgByTopic.sendDelayedAlarm(alarmTask, 1);
     }
 
 }

mh-quartz/src/main/java/com/mh/quartz/task/PushDataToWechatTask.java

@@ -52,11 +52,11 @@ public class PushDataToWechatTask {
             return;
         }
         alarmRecords1.forEach(alarmRecords2 -> {
+            // 每个小时推送两次
+            if (alarmRecords2.getSendNum() < 3) {
                 // 查询需要推送的微信用户
                 List<Map<String, Object>> wechatUserList = wechatService.queryWechatUser(0);
-            if (null == wechatUserList || wechatUserList.isEmpty()) {
+                if (null != wechatUserList && !wechatUserList.isEmpty()) {
-                log.info("没有查询到微信用户");
-            }
                     // 开始推送数据
                     for (Map<String, Object> map : wechatUserList) {
                         PushMsgEntity pushMsgEntity = new PushMsgEntity();
@@ -86,6 +86,8 @@ public class PushDataToWechatTask {
                             }
                         }
                     }
+                }
+            }
         });
     }
 
@@ -129,6 +131,7 @@ public class PushDataToWechatTask {
 
     /**
      * 推送设备异常告警提醒
+     *
      * @param map
      * @param pushMsgEntity
      * @param alarmRecords2

mh-system/src/main/java/com/mh/system/mapper/operation/AlarmRecordsMapper.java

@@ -15,6 +15,6 @@ import org.apache.ibatis.annotations.Update;
 @Mapper
 public interface AlarmRecordsMapper extends BaseMapper<AlarmRecords> {
 
-    @Update("update alarm_records set is_send = #{isSend} where id = #{id} ")
+    @Update("update alarm_records set is_send = #{isSend}, send_num = send_num + 1 where id = #{id} ")
     void updateIsSendById(String id, int isSend);
 }

mh-system/src/main/java/com/mh/system/service/operation/IAlarmRecordsService.java

@@ -26,4 +26,8 @@ public interface IAlarmRecordsService {
     int deleteAlarmRecordsByIds(String[] acdIds);
 
     void updateAlarmRecordById(String id, int i);
+
+    String createAlarmTask();
+
+    void insertOrUpdateAlarmRecord(String cpmId);
 }

mh-system/src/main/java/com/mh/system/service/operation/impl/AlarmRecordsServiceImpl.java

@@ -1,18 +1,31 @@
 package com.mh.system.service.operation.impl;
 
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.mh.common.core.domain.entity.AlarmCode;
 import com.mh.common.core.domain.entity.AlarmRecords;
+import com.mh.common.core.domain.entity.AlarmRules;
+import com.mh.common.core.domain.entity.CollectionParamsManage;
+import com.mh.common.core.redis.RedisCache;
+import com.mh.common.utils.BigDecimalUtils;
+import com.mh.common.utils.DateUtils;
 import com.mh.common.utils.StringUtils;
 import com.mh.system.mapper.operation.AlarmRecordsMapper;
+import com.mh.system.service.device.ICollectionParamsManageService;
+import com.mh.system.service.operation.IAlarmCodeService;
 import com.mh.system.service.operation.IAlarmRecordsService;
+import com.mh.system.service.operation.IAlarmRulesService;
 import jakarta.annotation.Resource;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.math.BigDecimal;
+import java.math.RoundingMode;
 import java.text.SimpleDateFormat;
 import java.time.LocalDate;
-import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
+import java.util.Date;
 import java.util.List;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @author LJF
@@ -27,6 +40,18 @@ public class AlarmRecordsServiceImpl implements IAlarmRecordsService {
     @Resource
     private AlarmRecordsMapper alarmRecordsMapper;
 
+    @Resource
+    private IAlarmRulesService alarmRulesService;
+
+    @Resource
+    private IAlarmCodeService alarmCodeService;
+
+    @Resource
+    private ICollectionParamsManageService collectionParamsManageService;
+
+    @Autowired
+    private RedisCache redisTemplate;
+
     @Override
     public AlarmRecords selectIsExist(AlarmRecords alarmRecords) {
         QueryWrapper<AlarmRecords> queryWrapper = new QueryWrapper<>();
@@ -137,4 +162,112 @@ public class AlarmRecordsServiceImpl implements IAlarmRecordsService {
     public void updateAlarmRecordById(String id, int isSend) {
         alarmRecordsMapper.updateIsSendById(id, isSend);
     }
+
+    @Override
+    public void insertOrUpdateAlarmRecord(String cpmId) {
+        // 查询仪表报警规则记录，查看哪些规则启用了
+        List<AlarmRules> alarmRules = alarmRulesService.selectAlarmRulesListByStatus(0);
+        // 循环查询报警规则，判断是否满足报警条件
+        for (AlarmRules alarmRule : alarmRules) {
+            if (!alarmRule.getCpmId().equals(cpmId)) {
+                continue;
+            }
+            // 判断报警类型
+            if ("0".equals(alarmRule.getAlarmType())) {
+                // 当前是越限事件
+                // 查询事件类型查询对应的报警模板内容
+                AlarmCode alarmCode = alarmCodeService.selectAlarmCodeByAlarmType(alarmRule.getEventType());
+                // 获取当前采集参数值
+                CollectionParamsManage collectionParamsManage = collectionParamsManageService.selectCollectionParamsManageById(alarmRule.getCpmId());
+                // 判断当前值是否是当前事件
+                if (null == collectionParamsManage) {
+                    continue;
+                }
+                AlarmRecords alarmRecords = new AlarmRecords();
+                BigDecimal curValue = collectionParamsManage.getCurValue();
+                Date curTime = collectionParamsManage.getCurTime();
+                // 阈值
+                String threshold1 = alarmRule.getThreshold1();
+                // 设置Redis取消标记（有效期略大于延迟时间）
+                redisTemplate.deleteObject("alarm:cancel:" + collectionParamsManage.getId());
+                if (alarmRule.getTimePeriodSet() == 0 && DateUtils.isSameDay(curTime, new Date())) {
+                    // 执行相关操作
+                    insertOrUpdateRecord(alarmRule, curValue, threshold1, alarmCode, alarmRecords, collectionParamsManage);
+                } else if (alarmRule.getTimePeriodSet() == 1
+                        && DateUtils.isSameDay(collectionParamsManage.getCurTime(), new Date())
+                        && DateUtils.isCurrentTimeInRange(alarmRule.getBeginTime(), alarmRule.getEndTime(), curTime)
+                ) {
+                    // 执行相关操作
+                    insertOrUpdateRecord(alarmRule, curValue, threshold1, alarmCode, alarmRecords, collectionParamsManage);
+                }
+            }
+        }
+    }
+
+    @Override
+    public String createAlarmTask() {
+        // 查询仪表报警规则记录，查看哪些规则启用了
+        List<AlarmRules> alarmRules = alarmRulesService.selectAlarmRulesListByStatus(0);
+        // 循环查询报警规则，判断是否满足报警条件
+        for (AlarmRules alarmRule : alarmRules) {
+            // 判断报警类型
+            if ("0".equals(alarmRule.getAlarmType())) {
+                // 当前是越限事件
+                // 获取当前采集参数值
+                CollectionParamsManage collectionParamsManage = collectionParamsManageService.selectCollectionParamsManageById(alarmRule.getCpmId());
+                // 判断当前值是否是当前事件
+                if (null == collectionParamsManage) {
+                    continue;
+                }
+                AlarmRecords alarmRecords = new AlarmRecords();
+                BigDecimal curValue = collectionParamsManage.getCurValue();
+                Date curTime = collectionParamsManage.getCurTime();
+                // 阈值
+                String threshold1 = alarmRule.getThreshold1();
+                if (curValue.compareTo(BigDecimal.ZERO) == 0 || curValue.compareTo(new BigDecimal(threshold1)) < 0) {
+                    // 发送到延迟队列
+                    return collectionParamsManage.getId();
+                } else {
+                    // 设置Redis取消标记（有效期略大于延迟时间）
+                    redisTemplate.setCacheObject(
+                            "alarm:cancel:" + collectionParamsManage.getId(),
+                            "true",
+                            2, TimeUnit.MINUTES  // 例如：延迟5分钟，设置10分钟过期
+                    );
+                }
+            }
+        }
+        return null;
+    }
+
+    private void insertOrUpdateRecord(AlarmRules alarmRule, BigDecimal curValue, String threshold1, AlarmCode alarmCode, AlarmRecords alarmRecords, CollectionParamsManage collectionParamsManage) {
+        boolean compare = BigDecimalUtils.compare(alarmRule.getCondition1(), curValue, new BigDecimal(threshold1));
+        if (compare) {
+            // 创建报警记录
+            String content = alarmCode.getMsgContent();
+            content = content.replace("#{curValue}", curValue.setScale(1, RoundingMode.HALF_UP).toString());
+            content = content.replace("#{setValue}", alarmRule.getCondition1() + threshold1);
+            alarmRecords.setContent(content);
+            alarmRecords.setAlarmType(alarmRule.getAlarmType());
+            alarmRecords.setEventType(alarmRule.getEventType());
+            alarmRecords.setAlarmLevel(alarmRule.getAlarmLevel());
+            alarmRecords.setLedgerId(alarmRule.getLedgerId());
+            alarmRecords.setCpmId(alarmRule.getCpmId());
+            alarmRecords.setDeviceName(alarmRule.getDeviceName());
+            alarmRecords.setCpmName(alarmRule.getCpmName());
+            alarmRecords.setCreateTime(collectionParamsManage.getCurTime());
+            // 判断报警记录是否已经存在
+            AlarmRecords isExits = selectIsExist(alarmRecords);
+            if (isExits == null) {
+                insertAlarmRecord(alarmRecords);
+            } else {
+                // 更新报警记录
+                isExits.setContent(content);
+                isExits.setCreateTime(collectionParamsManage.getCurTime());
+                isExits.setStatus(0);
+                isExits.setIsSend(0);
+                updateAlarmRecord(isExits);
+            }
+        }
+    }
 }