第二版,后台bug

src/main/java/com/care/CareSpringStart.java

@@ -1,6 +1,6 @@
 package com.care;
 
-import com.care.common.init.InitRunner;
+import com.care.mqtt.init.MqttInitRunner;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
@@ -38,8 +38,8 @@ public class CareSpringStart {
     }
 
     @Bean
-    public InitRunner mqttInit() {
-        return new InitRunner();
+    public MqttInitRunner mqttInit() {
+        return new MqttInitRunner();
     }
 }
 

src/main/java/com/care/common/cache/RedisUtil.java

@@ -570,4 +570,36 @@ public class RedisUtil {
             redisTemplate.delete(key);
         }
     }
-}
+
+    public boolean zSetAdd(String key, Object value, long score) {
+        try {
+            boolean re =  redisTemplate.opsForZSet().add(key, value, score);
+            return re;
+         } catch(Exception e) {
+            e.printStackTrace();
+            return false;
+        }
+    }
+
+    public Set zSetRangeByScore(String key, double min, double max) {
+        try {
+            Set value = redisTemplate.opsForZSet().rangeByScore(key, min, max);
+            return value;
+        } catch(Exception e) {
+            e.printStackTrace();
+            return null;
+        }
+    }
+
+    public Long zSetRemoveRangeByScore(String key, double min, double max) {
+        try {
+            Long value = redisTemplate.opsForZSet().removeRangeByScore(key, min, max);
+            return value;
+         } catch(Exception e) {
+            e.printStackTrace();
+            return null;
+        }
+
+    }
+
+}

src/main/java/com/care/common/constant/Constants.java

@@ -16,4 +16,8 @@ public interface Constants {
      */
     String LOGIN_TOKEN_KEY = "token";
 
+    /**
+     * Mqtt 延迟消息KEY
+     */
+    String MQTT_DELAY_MSG_KEY = "mqttDelayMsgKey";
 }

src/main/java/com/care/common/init/InitRunner.java

@@ -1,8 +1,8 @@
-package com.care.common.init;
+package com.care.mqtt.init;
 
 
 import com.care.mqtt.service.MqttConnectorPoolService;
-import com.care.mqtt.service.MqttThreadPoolService;
+import com.care.mqtt.service.MqttMsgRedisService;
 import com.care.common.util.CommonConfUtil;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
@@ -13,14 +13,14 @@ import org.springframework.core.annotation.Order;
 
 
 @Order(1)
-public class InitRunner implements CommandLineRunner {
+public class MqttInitRunner implements CommandLineRunner {
 
-    private static final Logger logger = LogManager.getLogger(InitRunner.class);
+    private static final Logger logger = LogManager.getLogger(MqttInitRunner.class);
 
     @Autowired
     private MqttConnectorPoolService mqttConnectorPoolService;
     @Autowired
-    private MqttThreadPoolService mqttThreadPoolService;
+    private MqttMsgRedisService mqttMsgRedisService;
 
     @Override
     public void run(String... args) throws Exception {
@@ -28,7 +28,7 @@ public class InitRunner implements CommandLineRunner {
         if ("1".equals(mqttOn)){
             logger.info("初始化MQTTstart ...........................");
             mqttConnectorPoolService.init();
-            mqttThreadPoolService.init();
+            mqttMsgRedisService.init();
             logger.info("初始化MQTTend ...........................");
         }
     }

src/main/java/com/care/mqtt/schedule/MqttMsgSchedule.java

@@ -0,0 +1,38 @@
+package com.care.mqtt.schedule;
+
+
+
+import com.care.mqtt.service.MqttMsgRedisService;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.Scheduled;
+
+
+
+/** 手动外呼话单调度
+ * @author 许明
+ * @version 2.7.0.0 创建于 2018/7/16
+ **/
+@Configuration
+@EnableScheduling
+public class MqttMsgSchedule {
+    private static final Logger logger = LogManager.getLogger(MqttMsgSchedule.class);
+   @Autowired
+    private MqttMsgRedisService mqttMsgRedisService;
+    /**
+     * 5 秒钟调用一次
+     */
+    @Scheduled(cron = "0/5 * * * * ?")
+    public void rotationRedis() {
+        try {
+            mqttMsgRedisService.rotationRedis();
+        } catch (Exception e) {
+            logger.error("轮训redis出错：{}", e.getMessage());
+        }
+    }
+
+}

src/main/java/com/care/mqtt/service/MqttThreadPoolService.java

@@ -4,32 +4,36 @@ package com.care.mqtt.service;
 import cn.hutool.core.collection.CollUtil;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.care.bigscreen.service.BigScreenService;
+import com.care.common.cache.RedisUtil;
+import com.care.common.constant.Constants;
 import com.care.common.entity.CareDevice;
 import com.care.common.entity.CareEventOrder;
 import com.care.common.entity.CareMqttMsg;
-import com.care.common.enums.*;
+import com.care.common.enums.MqttMsgStatusEnum;
+import com.care.common.enums.OrderStatusEnum;
+import com.care.common.enums.OrderTypeEnum;
 import com.care.common.service.CareDeviceService;
-
 import com.care.common.service.CareEventOrderService;
 import com.care.common.service.CareMqttMsgService;
+import org.apache.commons.collections.CollectionUtils;
 import org.springframework.stereotype.Service;
 
-
 import javax.annotation.Resource;
 import java.util.Date;
 import java.util.List;
-
+import java.util.Set;
 import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
 
 /**
+ * 使用redis轮训方案实现延迟触发
  * @Author: stw
  * @Date: 2021/8/14
  * @Desc:
  */
 @Service
-public class MqttThreadPoolService {
+public class MqttMsgRedisService {
     @Resource
     private CareDeviceService careDeviceService;
     @Resource
@@ -41,25 +45,13 @@ public class MqttThreadPoolService {
     @Resource
     private BigScreenService bigScreenService;
     @Resource
-    private ScheduledThreadPoolExecutor scheduledThreadPool;
+    private RedisUtil redisUtil;
 
-    public ScheduledThreadPoolExecutor getScheduledThreadPool() {
-        return scheduledThreadPool;
-    }
 
     /**
-     * 读取数据库mqtt msg表，把有延迟标记、未处理的消息生成调度任务放入线程池
+     * 读取数据库mqtt msg表，把有延迟标记、未处理的消息生成调redis结构放入redis
      */
     public void init() {
-        QueryWrapper<CareDevice> queryWrapper = new QueryWrapper();
-        queryWrapper.lambda().eq(CareDevice::getActiveStatus, DeviceActiveStatusEnum.ACTIVE.getValue());
-        int devCount = careDeviceService.count(queryWrapper);
-
-        if (devCount < 5) {
-            devCount = 5;
-        }
-        //创建一个定长线程池，支持定时任务执行——延迟执行
-         scheduledThreadPool = new ScheduledThreadPoolExecutor(devCount);
 
         //查询mqtt msg 表有延迟标记的未处理消息，放入线程池,等待延时时间结束执行创建工单
         QueryWrapper<CareMqttMsg> queryWrapper2 = new QueryWrapper<>();
@@ -73,30 +65,58 @@ public class MqttThreadPoolService {
                 Long now = System.currentTimeMillis();
                 Long createTime = careMqttMsg.getCreateTime().getTime();
                 Long delayTime = careMqttMsg.getDelayTime() * 60 * 1000L;
-                Long diff = delayTime - (now - createTime);
-                if(diff <= 0) { //立即执行
+                Long execTime = createTime + delayTime;
+
+                if(execTime <= now) { //立即执行
                     exe(careMqttMsg);
                     continue;
                 }
-                addScheduled(careMqttMsg,diff.intValue());
+                addRedis(careMqttMsg,execTime);
             }
 
         }
     }
 
-    public void addScheduled(CareMqttMsg careMqttMsg,int delay){
 
-       //executor.setCorePoolSize(newPoolSize);//todo
 
-        //延迟 diff 毫秒执行
-        scheduledThreadPool.schedule(new Runnable() {
-            public void run() {
+    /**
+     * 把延迟消息放入redis
+     * @param careMqttMsg
+     * @param execTime
+     */
+    public void addRedis(CareMqttMsg careMqttMsg, long execTime){
+
+        String msg = String.valueOf(careMqttMsg.getId());
+        redisUtil.zSetAdd(Constants.MQTT_DELAY_MSG_KEY, msg, execTime);
+    }
+
+    /**
+     * 轮训redis
+     */
+    public void rotationRedis(){
+        long startTime = 0L;
+        long endTime = System.currentTimeMillis();
+
+        //查询符合时间条件下的集合
+        Set set = redisUtil.zSetRangeByScore(Constants.MQTT_DELAY_MSG_KEY, startTime, endTime);
+
+        if (CollUtil.isNotEmpty(set)) {
+            //具体业务操作
+            set.forEach(item -> {
+                CareMqttMsg careMqttMsg = careMqttMsgService.getById((Long)item);
                 exe(careMqttMsg);
-            }
-        }, delay, TimeUnit.MILLISECONDS);
+            });
 
+            //移除集合
+            redisUtil.zSetRemoveRangeByScore(Constants.MQTT_DELAY_MSG_KEY, startTime, endTime);
+        }
     }
 
+    /**
+     *  执行创建工单动作
+     * @param careMqttMsg
+     */
+
     private void exe(CareMqttMsg careMqttMsg){
         String status = this.careMqttMsgService.getById(careMqttMsg).getStatus();
         if(!MqttMsgStatusEnum.NO_HANDLE.getValue().equals(status)){ //不是未处理，不执行。
@@ -105,8 +125,8 @@ public class MqttThreadPoolService {
 
         CareDevice careDevice = careDeviceService.getById(careMqttMsg.getDevId());
         CareEventOrder order = mqttMsgService.getNewOrder(careDevice);
-        order.setCreateTime(new Date());
-        order.setModifyTime(new Date());
+        order.setCreateTime(careMqttMsg.getCreateTime());
+        order.setModifyTime(careMqttMsg.getCreateTime());
         order.setStatus(OrderStatusEnum.TODO.getValue());
         if ("FallDown".equals(careMqttMsg.getType())) { //跌到
             order.setOrderType(OrderTypeEnum.DI_EDAO.getValue());

src/main/java/com/care/mqtt/service/MqttMsgService.java

@@ -48,7 +48,7 @@ public class MqttMsgService {
     @Resource
     private CareEventOrderChambService careEventOrderChambService;
     @Resource
-    private MqttThreadPoolService MqttThreadPoolService;
+    private MqttMsgRedisService mqttMsgRedisService;
     @Resource
     private CareRoomSensorService careRoomSensorService;
     @Resource
@@ -252,7 +252,7 @@ public class MqttMsgService {
                         this.careMqttMsgService.save(careMqttMsg);
 
                         //进行调度处理
-                        MqttThreadPoolService.addScheduled(careMqttMsg,careMqttMsg.getDelayTime() * 60 * 1000);
+                        mqttMsgRedisService.addRedis(careMqttMsg,date.getTime() + careMqttMsg.getDelayTime() * 60 * 1000);
                     } else { //取消： 更新mqtt消息为已取消，等待执行的调度到时会取消执行
                         careMqttMsg.setStatus(MqttMsgStatusEnum.HANDLED.getValue());
                         this.careMqttMsgService.save(careMqttMsg);