lixing 4 年 前
コミット
34995db47d

+ 6 - 4
src/main/java/com/persagy/client/GroupNettyClientHandler.java

@@ -20,6 +20,7 @@ import io.netty.handler.timeout.IdleStateEvent;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
 
+import javax.transaction.Transactional;
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.TimeUnit;
@@ -119,7 +120,8 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
         }
     }
 
-    private void handlerMsg(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
+    @Transactional
+    public void handlerMsg(ChannelHandlerContext channelHandlerContext, Object msg) throws Exception {
         if (StringUtil.isJSONObject((String) msg)) {
             NettyMessage message = StringUtil.tranferItemToDTO((String) msg, NettyMessage.class);
             if (message.getOpCode() == 9) {
@@ -131,12 +133,12 @@ public class GroupNettyClientHandler extends ChannelInboundHandlerAdapter {
                         LockUtil.getInstance().lock.lock();
                         LockUtil.getInstance().setExecute(false);
                         //加个等待,保证正在执行的逻辑执行成功
-                        Thread.sleep(40000);
+                        Thread.sleep(4000);
                         alarmDefineService.listAllAlarmDefine(definesList);
                         LockUtil.getInstance().setExecute(true);
                         LockUtil.getInstance().condition.signalAll();
-                    } catch (InterruptedException e) {
-                        e.printStackTrace();
+                    } catch (Exception e) {
+                        log.error("全量获取报警定义发生异常", e);
                     } finally {
                         LockUtil.getInstance().lock.unlock();
                     }

+ 12 - 9
src/main/java/com/persagy/service/impl/AlarmHandleServiceImpl.java

@@ -24,6 +24,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 
+import javax.transaction.Transactional;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
@@ -72,6 +73,7 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
      * @version: V1.0
      */
     @Override
+    @Transactional
     public void handleIOTData(String msg) throws SchedulerException, InterruptedException {
         log.info("接收到采集值:[{}]", msg);
         String[] split = msg.split(";");
@@ -120,14 +122,14 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                             //报警产生值满足(这里的满足不考虑报警持续时间)
                             if (triggerResult) {
                                 log.info("满足报警条件");
-                                log.info("--" + alarmDefine.toString());
-                                log.info("--" + JSONObject.toJSONString(paramMap));
+//                                log.info("--" + alarmDefine.toString());
+//                                log.info("--" + JSONObject.toJSONString(paramMap));
                                 //报警的时候不考虑报警恢复,因为同时报警和报警恢复是不应该出现的
                                 handlerNowDataAlarm(alarmDefine, dataTime, paramMap);
                             } else {
                                 log.info("不满足报警条件");
-                                log.info("--" + alarmDefine.toString());
-                                log.info("--" + JSONObject.toJSONString(paramMap));
+//                                log.info("--" + alarmDefine.toString());
+//                                log.info("--" + JSONObject.toJSONString(paramMap));
                                 //当前数据正常
                                 handlerNowDataNormal(alarmDefine, dataTime, endResult, paramMap);
                             }
@@ -245,10 +247,10 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                 jobDataMap.put("defineId", defineId);
                 //恢复
                 jobDataMap.put("state", "2");
-                log.info(JSONObject.toJSONString(jobDataMap));
+                log.info("发送恢复消息的任务参数:{}", JSONObject.toJSONString(jobDataMap));
                 String jobName = defineId;
                 ExpireAlarmMessage em = new ExpireAlarmMessage();
-                //过期消息
+                //恢复消息
                 em.setType("1");
                 em.setStartTime(DateUtil.offsetMinute(new Date(), 3).toJdkDate());
                 em.setJobDataMap(jobDataMap);
@@ -370,13 +372,14 @@ public class AlarmHandleServiceImpl implements AlarmHandleService {
                 log.info("设定的持续时间:[{}]", condition.getTriggerUphold());
                 if (alarmLastTime >= condition.getTriggerUphold()) {
                     log.info("报警持续时间大于设定的持续时间,产生一条报警");
-                    // 发送创建报警记录的消息
-                    sendCreateAlarmRecordMessage(alarmDefine, dataTime, paramMap,
-                            DateUtils.localDateTime2Date(expireDateTime));
                     // 边缘端保存报警信息
                     upsertZktAlarmRecordWhenAlarmStart(alarmDefine);
                     // 报警定义状态更新为未处理
                     alarmDefineState.setState(AlarmDefineState.State.NOT_DEAL.getType());
+                    // 发送创建报警记录的消息
+                    sendCreateAlarmRecordMessage(alarmDefine, dataTime, paramMap,
+                            DateUtils.localDateTime2Date(expireDateTime));
+
                     //有过期时间,生成报警过期消息
                     if (expireDateTime != null) {
                         //过期时间

+ 77 - 22
src/main/java/com/persagy/service/impl/AlarmQuartzServiceImpl.java

@@ -1,6 +1,7 @@
 package com.persagy.service.impl;
 
 import cn.hutool.core.date.DateUtil;
+import com.alibaba.fastjson.JSONObject;
 import com.persagy.entity.ExpireAlarmMessage;
 import com.persagy.job.AlarmExpireJob;
 import com.persagy.job.ExpireAlarmQueue;
@@ -8,11 +9,11 @@ import com.persagy.service.AlarmQuartzService;
 import com.persagy.utils.DateUtils;
 import lombok.extern.slf4j.Slf4j;
 import org.quartz.*;
+import org.quartz.impl.matchers.GroupMatcher;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.PostConstruct;
 import java.util.*;
 
 import static org.quartz.SimpleScheduleBuilder.simpleSchedule;
@@ -38,45 +39,48 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
             System.out.println("--init--");
             //启动后先休息20秒,这样报警netty  websocket  quartz都启动了
             Thread.sleep(20000);
-            while (true){
+            while (true) {
                 ExpireAlarmMessage expireAlarmMessage = ExpireAlarmQueue.getExpireAlarmMessageQueue().consume();
-                log.info("剩余过期消息总数:{}",ExpireAlarmQueue.getExpireAlarmMessageQueue().size());
-                if("1".equals(expireAlarmMessage.getType())){
-                    addExpireJob(expireAlarmMessage.getStartTime(),expireAlarmMessage.getJobName(),expireAlarmMessage.getJobGroupName(),expireAlarmMessage.getJobDataMap());
-                }else if("2".equals(expireAlarmMessage.getType())){
-                    deleteExpireJob(expireAlarmMessage.getJobName(),expireAlarmMessage.getJobGroupName());
+                log.info("剩余过期消息总数:{}", ExpireAlarmQueue.getExpireAlarmMessageQueue().size());
+                if ("1".equals(expireAlarmMessage.getType())) {
+                    addExpireJob(expireAlarmMessage.getStartTime(), expireAlarmMessage.getJobName(), expireAlarmMessage.getJobGroupName(), expireAlarmMessage.getJobDataMap());
+                } else if ("2".equals(expireAlarmMessage.getType())) {
+                    deleteExpireJob(expireAlarmMessage.getJobName(), expireAlarmMessage.getJobGroupName());
                 }
             }
         } catch (Exception e) {
-            log.error("报警过期消息队列消费失败",e);
+            log.error("报警过期消息队列消费失败", e);
         }
     }
+
     /**
      * 添加定时器
-     * @param startTime 添加报警恢复定时任务
-     * @param jobName   job名称
-     * @param jobGroupName  job分组
-     * @param jobDataMap  触发器的数据(任务相关的数据都放在这里)
-     * @return 
+     *
+     * @param startTime    添加报警恢复定时任务
+     * @param jobName      job名称
+     * @param jobGroupName job分组
+     * @param jobDataMap   触发器的数据(任务相关的数据都放在这里)
+     * @return
      * @throws SchedulerException
      */
     @Override
     public String addExpireJob(Date startTime, String jobName, String jobGroupName, JobDataMap jobDataMap) throws SchedulerException {
-        if(startTime.before(new Date())){
+        if (startTime.before(new Date())) {
             log.error("执行时间【{}】为历史时间,继续添加", DateUtils.formatDate(startTime));
         }
         JobKey jobKey = new JobKey(jobName, jobGroupName);
+        log.info("恢复/过期任务jobKey: [{}]", jobKey);
         JobDetail jobDetail = JobBuilder.newJob(AlarmExpireJob.class).withIdentity(jobName, jobGroupName).requestRecovery().build();
         Trigger trigger = TriggerBuilder.newTrigger().startAt(startTime)
-                .withIdentity("trigger_" + jobName , jobName)
+                .withIdentity("trigger_" + jobName, jobGroupName)
                 .usingJobData(jobDataMap)
                 .withSchedule(simpleSchedule().withMisfireHandlingInstructionIgnoreMisfires()).build();
         //已经存在的job实例和触发器自动覆盖 job实 例 唯 一标识:
         HashSet<Trigger> triggerSet = new HashSet<Trigger>();
-        Date fireTime = trigger.getFireTimeAfter(DateUtil.offsetMinute(startTime,-1).toJdkDate());
-        if(fireTime != null){
-            log.info("执行时间为:【{}】,设置时间为【{}】",DateUtils.formatDate(fireTime),DateUtils.formatDate(startTime));
-        }else {
+        Date fireTime = trigger.getFireTimeAfter(DateUtil.offsetMinute(startTime, -1).toJdkDate());
+        if (fireTime != null) {
+            log.info("执行时间为:【{}】,设置时间为【{}】", DateUtils.formatDate(fireTime), DateUtils.formatDate(startTime));
+        } else {
             log.info("执行时间为 空!");
         }
         triggerSet.add(trigger);
@@ -84,14 +88,59 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
         return "success";
     }
 
+    /**
+     * @description: 获取当前所有定时任务的信息,调试时使用,上线请勿用
+     * @param:
+     * @return: java.util.List<java.util.Map < java.lang.String, java.lang.Object>>
+     * @exception:
+     * @author: lixing
+     * @company: Persagy Technology Co.,Ltd
+     * @since: 2021/1/14 下午1:15
+     * @version: V1.0
+     */
+    public List<Map<String, Object>> queryAllJob() {
+        List<Map<String, Object>> jobList = null;
+        try {
+            GroupMatcher<JobKey> matcher = GroupMatcher.anyJobGroup();
+            Set<JobKey> jobKeys = quartzScheduler.getJobKeys(matcher);
+            jobList = new ArrayList<Map<String, Object>>();
+            for (JobKey jobKey : jobKeys) {
+                List<? extends Trigger> triggers = quartzScheduler.getTriggersOfJob(jobKey);
+                for (Trigger trigger : triggers) {
+                    Map<String, Object> map = new HashMap<>();
+                    map.put("jobName", jobKey.getName());
+                    map.put("jobGroupName", jobKey.getGroup());
+                    map.put("description", "触发器:" + trigger.getKey());
+                    Trigger.TriggerState triggerState = quartzScheduler.getTriggerState(trigger.getKey());
+                    map.put("jobStatus", triggerState.name());
+                    if (trigger instanceof CronTrigger) {
+                        CronTrigger cronTrigger = (CronTrigger) trigger;
+                        String cronExpression = cronTrigger.getCronExpression();
+                        map.put("jobTime", cronExpression);
+                    } else if (trigger instanceof SimpleTrigger) {
+                        SimpleTrigger simpleTrigger = (SimpleTrigger) trigger;
+                        map.put("jobTime", simpleTrigger.getStartTime());
+                    }
+
+                    jobList.add(map);
+                }
+            }
+        } catch (SchedulerException e) {
+            log.error("获取当前定时任务发生异常", e);
+        }
+        log.info("当前所有定时任务数量:[{}]", jobList.size());
+        log.info("当前所有定时任务:[{}]", JSONObject.toJSONString(jobList));
+        return jobList;
+    }
+
 
     /**
+     * @param jobName:job名称
+     * @param jobGroupName:job分组名称
      * @description:删除定时任务
      * @exception:
      * @author: LuoGuangyi
      * @company: Persagy Technology Co.,Ltd
-     * @param jobName:job名称
-     * @param jobGroupName:job分组名称
      * @return: java.lang.String
      * @since: 2020/09/29 9:23
      * @version: V1.0
@@ -99,7 +148,13 @@ public class AlarmQuartzServiceImpl implements AlarmQuartzService {
     @Override
     public synchronized String deleteExpireJob(String jobName, String jobGroupName) throws SchedulerException {
         JobKey jobKey = new JobKey(jobName, jobGroupName);
-        quartzScheduler.deleteJob(jobKey);
+        boolean exists = quartzScheduler.checkExists(jobKey);
+        if (exists) {
+            log.info("移除过期任务:[{}]", jobKey);
+            quartzScheduler.deleteJob(jobKey);
+        } else {
+            log.info("要移除过期任务不存在!可能已经执行完成~");
+        }
         return "success!";
     }