延迟队列代码部分提交

soc-api/soc-api-system/src/main/java/com/xunmei/system/api/util/LogUtils.java

@@ -177,4 +177,9 @@ public class LogUtils {
      * fsu动环和报警主机传感器短信日志
      */
     public  static  final Logger SMS_NOTICE_LOG=LoggerFactory.getLogger("smsNoticeLog");
+
+    /**
+     * websocket消息重试日志
+     */
+    public  static  final Logger WS_MSG_RETRY_LOG=LoggerFactory.getLogger("wsMsgRetryLog");
 }

soc-common/soc-common-core/src/main/java/com/xunmei/common/core/thread/ThreadPoolConfig.java

@@ -17,7 +17,7 @@ public class ThreadPoolConfig implements AsyncConfigurer {
      * 项目共用线程池
      */
     public static final String SOC_EXECUTOR = "socExecutor";
-    public static final String MEDIATOR_EXECUTOR = "mediatorExecutor";
+    public static final String HOST_EXECUTOR = "mediatorExecutor";
 
 
     @Override
@@ -43,7 +43,7 @@ public class ThreadPoolConfig implements AsyncConfigurer {
         return executor;
     }
 
-    @Bean(MEDIATOR_EXECUTOR)
+    @Bean(HOST_EXECUTOR)
     @Primary
     public ThreadPoolTaskExecutor mediatorExecutor() {
         ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
@@ -52,7 +52,7 @@ public class ThreadPoolConfig implements AsyncConfigurer {
         executor.setCorePoolSize(10);
         executor.setMaxPoolSize(10);
         executor.setQueueCapacity(200);
-        executor.setThreadNamePrefix("soc-executor-");
+        executor.setThreadNamePrefix("host-executor-");
         //拒绝策略，由投递者执行
         executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
         //装饰器,用于异步线程内捕获异常

soc-common/soc-common-redis/src/main/java/com/xunmei/common/redis/enums/RedisDelayQueueEnum.java

@@ -0,0 +1,36 @@
+package com.xunmei.common.redis.enums;
+
+/**
+ * 延迟队列业务枚举
+ */
+
+
+import lombok.AllArgsConstructor;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+@Getter
+@NoArgsConstructor
+@AllArgsConstructor
+public enum RedisDelayQueueEnum {
+
+    WEBSOCKET_MSG_RETRY("WEBSOCKET_MSG_RETRY","websocket消息重试队列", "iotWebsocketMsgService"),
+
+    ;
+
+    /**
+     * 延迟队列 Redis Key
+     */
+    private String code;
+ 
+    /**
+     * 中文描述
+     */
+    private String name;
+ 
+    /**
+     * 延迟队列具体业务实现的 Bean
+     * 可通过 Spring 的上下文获取
+     */
+    private String beanId;
+ 
+}

soc-common/soc-common-redis/src/main/java/com/xunmei/common/redis/utils/RedisDelayedQueueUtil.java

@@ -0,0 +1,111 @@
+package com.xunmei.common.redis.utils;
+
+import com.xunmei.common.core.utils.SpringUtils;
+import lombok.NonNull;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.redisson.api.RBlockingDeque;
+import org.redisson.api.RDelayedQueue;
+import org.redisson.api.RedissonClient;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+
+public class RedisDelayedQueueUtil {
+
+    private static final RedissonClient CLIENT = SpringUtils.getBean(RedissonClient.class);
+
+
+    public static <T> void addDelayQueue(T value, LocalDateTime endTime, String queueCode) {
+        long seconds = Duration.between(LocalDateTime.now(), endTime).getSeconds();
+        if (seconds > 0) {
+            addDelayQueue(value, seconds, TimeUnit.SECONDS, queueCode);
+        }
+    }
+    public static <T> void addDelayQueue(T value, long delay, String queueCode) {
+        addDelayQueue(value, delay, TimeUnit.SECONDS, queueCode);
+    }
+    /**
+     * 添加延迟队列
+     * @param value 队列值
+     * @param delay 延迟时间
+     * @param timeUnit 时间单位
+     * @param queueCode 队列键
+     * @param <T> 泛型
+     */
+    public static <T> void addDelayQueue(T value, long delay, TimeUnit timeUnit, String queueCode){
+        try {
+            RBlockingDeque<Object> blockingDeque = CLIENT.getBlockingDeque(queueCode);
+            RDelayedQueue<Object> delayedQueue = CLIENT.getDelayedQueue(blockingDeque);
+            delayedQueue.offer(value, delay, timeUnit);
+            log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
+            //释放队列
+            delayedQueue.destroy();
+        } catch (Exception e) {
+            log.error("(添加延时队列失败) {}, value {}", e.getMessage(), value);
+            throw new RuntimeException("(添加延时队列失败)");
+        }
+    }
+
+
+    /**
+     * 获取延迟队列
+     * @param queueCode 队列主键
+     * @param <T> 泛型
+     * @return
+     * @throws InterruptedException
+     */
+    public static <T> T getDelayQueue(String queueCode) throws InterruptedException {
+        RBlockingDeque<T> blockingDeque = CLIENT.getBlockingDeque(queueCode);
+        //避免消息伪丢失（应用重启未消费），官网推荐
+        CLIENT.getDelayedQueue(blockingDeque) ;
+        return (T) blockingDeque.take();
+    }
+
+
+    public static boolean isContain(@NonNull Object o, @NonNull String queueCode) {
+        if (StringUtils.isBlank(queueCode)) {
+            return false;
+        }
+        try {
+            RBlockingDeque<Object> blockingDeque = CLIENT.getBlockingDeque(queueCode);
+            RDelayedQueue<Object> delayedQueue = CLIENT.getDelayedQueue(blockingDeque);
+            boolean flag = delayedQueue.contains(o);
+            if(flag){
+                log.info("(存在延时队列对应的值) 队列键：{}，队列值：{}", queueCode, o);
+            }
+            delayedQueue.destroy();
+            return flag;
+        } catch (Exception e) {
+            log.error("(判断存在延时队列异常) 队列键：{}，队列值：{},错误信息：{}",queueCode, o, e.getMessage());
+            throw new RuntimeException("(判断存在延时队列异常)");
+        }
+
+    }
+
+    public static boolean removeDelayedQueue(@NonNull Object o, @NonNull String queueCode) {
+        if (StringUtils.isBlank(queueCode)) {
+            return false;
+        }
+        try {
+            RBlockingDeque<Object> blockingDeque = CLIENT.getBlockingDeque(queueCode);
+            RDelayedQueue<Object> delayedQueue = CLIENT.getDelayedQueue(blockingDeque);
+            boolean flag = false;
+            if (delayedQueue.contains(o)) {
+                flag = delayedQueue.remove(o);
+            }
+            if(flag){
+                log.info("(删除延时队列保证唯一性) 队列键：{}，队列值：{}", queueCode, o);
+            }
+            delayedQueue.destroy();
+            return flag;
+        } catch (Exception e) {
+            log.error("(删除延时队列异常) 队列键：{}，队列值：{},错误信息：{}",queueCode, o, e.getMessage());
+            throw new RuntimeException("(删除延时队列异常)");
+        }
+
+    }
+}

soc-modules/soc-modules-host/src/main/java/com/xunmei/host/protection/service/impl/IotAlarmSubSystemServiceImpl.java

@@ -13,7 +13,6 @@ import com.xunmei.common.core.domain.iot.domain.IotAlarmSubsystem;
 import com.xunmei.common.core.domain.iot.domain.IotAlarmSubsystemLog;
 import com.xunmei.common.core.domain.iot.domain.IotDevice;
 import com.xunmei.common.core.enums.CategoryDataEnum;
-import com.xunmei.common.core.thread.ThreadPoolConfig;
 import com.xunmei.common.core.util.BeanHelper;
 import com.xunmei.common.core.utils.IDHelper;
 import com.xunmei.common.redis.utils.RedisUtils;
@@ -41,9 +40,6 @@ import com.xunmei.system.api.util.LogUtils;
 import com.xunmei.system.api.vo.SysOrgVO;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -53,7 +49,6 @@ import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
 
 
 /**
@@ -79,9 +74,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     private IotAlarmDefenceAreaService defenceAreService;
     @Resource
     private NorthErrorService northErrorService;
-    @Autowired
-    @Qualifier(ThreadPoolConfig.MEDIATOR_EXECUTOR)
-    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
+
     @Resource
     private IotAlarmSubsystemMapper iotAlarmSubsystemMapper;
 
@@ -198,10 +191,10 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         protection.setStatusUpdatorName(null);
         this.updateById(protection);
         //临时换文件处理,等下次打完整包的时候删除这段代码
-        LambdaUpdateWrapper<IotAlarmSubsystem> updateWrapper=new LambdaUpdateWrapper<>();
-        updateWrapper.eq(IotAlarmSubsystem::getId,protection.getId());
-        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorId,null);
-        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorName,null);
+        LambdaUpdateWrapper<IotAlarmSubsystem> updateWrapper = new LambdaUpdateWrapper<>();
+        updateWrapper.eq(IotAlarmSubsystem::getId, protection.getId());
+        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorId, null);
+        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorName, null);
         this.update(updateWrapper);
 
         this.northStatisticsSyncService.saveOrUpdateBusinessCountByDateAndDataType(DateUtil.today(), DataType.protectionStatusCount.getIndex(), 1L);
@@ -234,9 +227,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         }
         LogUtils.BASE_INFO_DEFENCEAREA.info("获取全部的报警主机子系统及防区信息,开始数据处理界面！");
         this.northStatisticsSyncService.saveOrUpdateBaseCountByDataType(DataType.protectionCount.getIndex(), defenceareaList.size(), false);
-        CompletableFuture.runAsync(() -> {
-            this.saveAndUpdate(defenceareaList, msgId);
-        }, threadPoolTaskExecutor);
+        this.saveAndUpdate(defenceareaList, msgId);
         return ReceiveErrorDto.success();
     }
 
@@ -253,9 +244,9 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     public IotAlarmSubsystem getByDeviceNameAndSubSystemId(String iotToken, String deviceName, String subSystemId) {
         String key = DeviceCacheEnum.IOT_SUB_SYSTEM.getCode() + iotToken + "_" + deviceName + "_" + subSystemId;
         IotAlarmSubsystem info = RedisUtils.getCacheObject(key);
-        if (info == null){
+        if (info == null) {
             IotAlarmSubsystem subsystem = iotAlarmSubsystemMapper.getByDeviceNameAndSubSystemId(iotToken, deviceName, String.valueOf(subSystemId));
-            if (subsystem != null){
+            if (subsystem != null) {
                 updateSubSystemCache(subsystem);
                 return subsystem;
             }
@@ -266,7 +257,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void saveSubSystemInfos(List<IotAlarmSubsystem> list) {
-        if (!list.isEmpty()){
+        if (!list.isEmpty()) {
             saveBatch(list);
             for (IotAlarmSubsystem data : list) {
                 updateSubSystemCache(data);
@@ -277,7 +268,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void updateSubSystemInfos(List<IotAlarmSubsystem> list) {
-        if (!list.isEmpty()){
+        if (!list.isEmpty()) {
             updateBatchById(list);
             for (IotAlarmSubsystem data : list) {
                 updateSubSystemCache(data);
@@ -290,7 +281,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         return baseMapper.findSubSystem(orgId, productName, deviceName, deviceId);
     }
 
-    private void updateSubSystemCache(IotAlarmSubsystem subsystem){
+    private void updateSubSystemCache(IotAlarmSubsystem subsystem) {
         String key = DeviceCacheEnum.IOT_SUB_SYSTEM.getCode() + subsystem.getIotToken() + "_" + subsystem.getAlarmHostCode() + "_" + subsystem.getCode();
         RedisUtils.setCacheObject(key, subsystem, Duration.ofMillis(1000 * 60 * 60));
     }

soc-modules/soc-modules-host/src/main/java/com/xunmei/host/websocket/redis/delay/RedisDelayQueueHandle.java

@@ -0,0 +1,10 @@
+package com.xunmei.host.websocket.redis.delay;
+
+/**
+ * 延迟队列执行器
+ */
+public interface RedisDelayQueueHandle<T> {
+ 
+    void execute(T t);
+ 
+}

soc-modules/soc-modules-host/src/main/java/com/xunmei/host/websocket/redis/delay/RedisDelayQueueRunner.java

@@ -0,0 +1,54 @@
+package com.xunmei.host.websocket.redis.delay;
+
+import cn.hutool.extra.spring.SpringUtil;
+import com.xunmei.common.core.thread.ThreadPoolConfig;
+import com.xunmei.common.redis.enums.RedisDelayQueueEnum;
+import com.xunmei.common.redis.utils.RedisDelayedQueueUtil;
+import com.xunmei.system.api.util.LogUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.boot.CommandLineRunner;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Component
+public class RedisDelayQueueRunner implements CommandLineRunner {
+
+
+    @Autowired
+    @Qualifier(ThreadPoolConfig.HOST_EXECUTOR)
+    private ThreadPoolTaskExecutor ptask;
+
+    @Override
+    public void run(String... args) throws Exception {
+        Thread thread = new Thread(()->{
+            while (!Thread.currentThread().isInterrupted()) {
+                try {
+                    RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
+                    for (RedisDelayQueueEnum queueEnum : queueEnums) {
+                        Object value = RedisDelayedQueueUtil.getDelayQueue(queueEnum.getCode());
+                        if (value != null) {
+                            LogUtils.WS_MSG_RETRY_LOG.info("{} 延迟队列有数据，开始处理", queueEnum.getName());
+                            RedisDelayQueueHandle<Object> redisDelayQueueHandle = (RedisDelayQueueHandle<Object>) SpringUtil.getBean(queueEnum.getBeanId());
+                            ptask.execute(() -> {
+                                redisDelayQueueHandle.execute(value);
+                            });
+                        }
+                    }
+                    TimeUnit.MILLISECONDS.sleep(500);
+                } catch (InterruptedException e) {
+                    LogUtils.WS_MSG_RETRY_LOG.error("(Redission延迟队列监测异常中断) {}", e.getMessage());
+                }
+            }
+        });
+        thread.setDaemon(true);
+        thread.setName("websocketMsg-delay—retry");
+
+        thread.start();
+        LogUtils.WS_MSG_RETRY_LOG.info("Redission延迟队列监测启动!");
+    }
+}

soc-modules/soc-modules-host/src/main/resources/logback.xml

@@ -34,7 +34,22 @@
 
 
 
-
+    <!-- websocket消息重试日志  -->
+    <appender name="wsMsgRetryLog" class="ch.qos.logback.core.rolling.RollingFileAppender">
+        <rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
+            <maxFileSize>${max.file.size}</maxFileSize>
+            <!--日志文件输出的文件名-->
+            <FileNamePattern>${LOG_HOME}/wsMsgRetryLog/%d{yyyy-MM-dd}-%i.log</FileNamePattern>
+            <!--日志文件保留天数-->
+            <maxHistory>${max.history}</maxHistory>
+            <totalSizeCap>${total.size.cap}</totalSizeCap>
+            <cleanHistoryOnStart>true</cleanHistoryOnStart>
+        </rollingPolicy>
+        <encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
+            <!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符-->
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+        </encoder>
+    </appender>
 
     <!-- websocket日志  -->
     <appender name="websocketMsgLog" class="ch.qos.logback.core.rolling.RollingFileAppender">
@@ -246,6 +261,10 @@
         <!--上面appender元素的name值。为了和logger的name属性做区分，我故意写的不一样-->
         <appender-ref ref="statusSensorLog"/>
     </logger>
+    <logger name="wsMsgRetryLog" additivity="false" level="info">
+        <!--上面appender元素的name值。为了和logger的name属性做区分，我故意写的不一样-->
+        <appender-ref ref="wsMsgRetryLog"/>
+    </logger>
 
     <!-- 日志输出级别 -->
     <root level="INFO">

soc-modules/soc-modules-mediator/src/main/java/com/xunmei/mediator/api/protection/service/impl/IotAlarmSubSystemServiceImpl.java

@@ -13,7 +13,6 @@ import com.xunmei.common.core.domain.iot.domain.IotAlarmSubsystem;
 import com.xunmei.common.core.domain.iot.domain.IotAlarmSubsystemLog;
 import com.xunmei.common.core.domain.iot.domain.IotDevice;
 import com.xunmei.common.core.enums.CategoryDataEnum;
-import com.xunmei.common.core.thread.ThreadPoolConfig;
 import com.xunmei.common.core.util.BeanHelper;
 import com.xunmei.common.core.utils.IDHelper;
 import com.xunmei.common.redis.utils.RedisUtils;
@@ -41,9 +40,6 @@ import com.xunmei.system.api.util.LogUtils;
 import com.xunmei.system.api.vo.SysOrgVO;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -53,7 +49,6 @@ import java.time.Duration;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
-import java.util.concurrent.CompletableFuture;
 
 
 /**
@@ -79,9 +74,6 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     private IotAlarmDefenceAreaService defenceAreService;
     @Resource
     private NorthErrorService northErrorService;
-    @Autowired
-    @Qualifier(ThreadPoolConfig.MEDIATOR_EXECUTOR)
-    private ThreadPoolTaskExecutor threadPoolTaskExecutor;
     @Resource
     private IotAlarmSubsystemMapper iotAlarmSubsystemMapper;
 
@@ -198,10 +190,10 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         protection.setStatusUpdatorName(null);
         this.updateById(protection);
         //临时换文件处理,等下次打完整包的时候删除这段代码
-        LambdaUpdateWrapper<IotAlarmSubsystem> updateWrapper=new LambdaUpdateWrapper<>();
-        updateWrapper.eq(IotAlarmSubsystem::getId,protection.getId());
-        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorId,null);
-        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorName,null);
+        LambdaUpdateWrapper<IotAlarmSubsystem> updateWrapper = new LambdaUpdateWrapper<>();
+        updateWrapper.eq(IotAlarmSubsystem::getId, protection.getId());
+        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorId, null);
+        updateWrapper.set(IotAlarmSubsystem::getStatusUpdatorName, null);
         this.update(updateWrapper);
 
         this.northStatisticsSyncService.saveOrUpdateBusinessCountByDateAndDataType(DateUtil.today(), DataType.protectionStatusCount.getIndex(), 1L);
@@ -234,9 +226,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         }
         LogUtils.BASE_INFO_DEFENCEAREA.info("获取全部的报警主机子系统及防区信息,开始数据处理界面！");
         this.northStatisticsSyncService.saveOrUpdateBaseCountByDataType(DataType.protectionCount.getIndex(), defenceareaList.size(), false);
-        CompletableFuture.runAsync(() -> {
-            this.saveAndUpdate(defenceareaList, msgId);
-        }, threadPoolTaskExecutor);
+        this.saveAndUpdate(defenceareaList, msgId);
         return ReceiveErrorDto.success();
     }
 
@@ -253,9 +243,9 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     public IotAlarmSubsystem getByDeviceNameAndSubSystemId(String iotToken, String deviceName, Integer subSystemId) {
         String key = DeviceCacheEnum.IOT_SUB_SYSTEM.getCode() + iotToken + "_" + deviceName + "_" + subSystemId;
         IotAlarmSubsystem info = RedisUtils.getCacheObject(key);
-        if (info == null){
+        if (info == null) {
             IotAlarmSubsystem subsystem = iotAlarmSubsystemMapper.getByDeviceNameAndSubSystemId(iotToken, deviceName, String.valueOf(subSystemId));
-            if (subsystem != null){
+            if (subsystem != null) {
                 updateSubSystemCache(subsystem);
                 return subsystem;
             }
@@ -266,7 +256,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void saveSubSystemInfos(List<IotAlarmSubsystem> list) {
-        if (!list.isEmpty()){
+        if (!list.isEmpty()) {
             saveBatch(list);
             for (IotAlarmSubsystem data : list) {
                 updateSubSystemCache(data);
@@ -277,7 +267,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
     @Override
     @Transactional(rollbackFor = Exception.class)
     public void updateSubSystemInfos(List<IotAlarmSubsystem> list) {
-        if (!list.isEmpty()){
+        if (!list.isEmpty()) {
             updateBatchById(list);
             for (IotAlarmSubsystem data : list) {
                 updateSubSystemCache(data);
@@ -285,7 +275,7 @@ public class IotAlarmSubSystemServiceImpl extends ServiceImpl<IotAlarmSubsystemM
         }
     }
 
-    private void updateSubSystemCache(IotAlarmSubsystem subsystem){
+    private void updateSubSystemCache(IotAlarmSubsystem subsystem) {
         String key = DeviceCacheEnum.IOT_SUB_SYSTEM.getCode() + subsystem.getIotToken() + "_" + subsystem.getAlarmHostCode() + "_" + subsystem.getCode();
         RedisUtils.setCacheObject(key, subsystem, Duration.ofMillis(1000 * 60 * 60));
     }