优化设备信息缓存redis

doc/nacos-config/DEFAULT_GROUP/application-dev.yml

@@ -12,6 +12,8 @@ ribbon:
 
 # feign 配置
 feign:
+  hystrix:
+    enabled: true
   sentinel:
     enabled: true
   okhttp:

thinglinks-api/thinglinks-api-broker/pom.xml

@@ -18,11 +18,6 @@
             <artifactId>thinglinks-common-core</artifactId>
             <version>${thinglinks.version}</version>
         </dependency>
-
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

thinglinks-api/thinglinks-api-link/pom.xml

@@ -25,10 +25,5 @@
             <version>${thinglinks.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
-
     </dependencies>
 </project>
\ No newline at end of file

thinglinks-api/thinglinks-api-monitor/pom.xml

@@ -18,11 +18,6 @@
             <artifactId>thinglinks-common-core</artifactId>
             <version>${thinglinks.version}</version>
         </dependency>
-
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

thinglinks-api/thinglinks-api-tdengine/pom.xml

@@ -19,10 +19,6 @@
             <version>${thinglinks.version}</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
     </dependencies>
 
 </project>
\ No newline at end of file

thinglinks-auth/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 服务注册地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-common/thinglinks-common-core/pom.xml

@@ -107,6 +107,11 @@
             <artifactId>swagger-annotations</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
         <!-- https://mvnrepository.com/artifact/cn.hutool/hutool-all -->
         <dependency>
             <groupId>cn.hutool</groupId>

thinglinks-common/thinglinks-common-core/src/main/java/com/mqttsnet/thinglinks/common/core/constant/Constants.java

@@ -126,4 +126,19 @@ public class Constants
      * 文件上传类型
      */
     public static final String APPLICATION_OCTET_STREAM = "application/octet-stream";
+
+    /**
+     * 设备信息 cache key
+     */
+    public static final String DEVICE_RECORD_KEY = "device_record:";
+
+    /**
+     * 重复提交 cache key
+     */
+    public static final String RESUBMIT_URL_KEY = "resubmit_url:";
+
+    /**
+     * 分布式锁机制 cache key
+     */
+    public static final String SET_NX = "setnx:";
 }

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/enums/DeviceConnectStatus.java → thinglinks-common/thinglinks-common-core/src/main/java/com/mqttsnet/thinglinks/common/core/enums/DeviceConnectStatus.java

-package com.mqttsnet.thinglinks.link.common.enums;
+package com.mqttsnet.thinglinks.common.core.enums;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/enums/DeviceStatus.java → thinglinks-common/thinglinks-common-core/src/main/java/com/mqttsnet/thinglinks/common/core/enums/DeviceStatus.java

-package com.mqttsnet.thinglinks.link.common.enums;
+package com.mqttsnet.thinglinks.common.core.enums;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/enums/DeviceType.java → thinglinks-common/thinglinks-common-core/src/main/java/com/mqttsnet/thinglinks/common/core/enums/DeviceType.java

-package com.mqttsnet.thinglinks.link.common.enums;
+package com.mqttsnet.thinglinks.common.core.enums;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/enums/ProtocolType.java → thinglinks-common/thinglinks-common-core/src/main/java/com/mqttsnet/thinglinks/common/core/enums/ProtocolType.java

-package com.mqttsnet.thinglinks.link.common.enums;
+package com.mqttsnet.thinglinks.common.core.enums;
 
 import lombok.AllArgsConstructor;
 import lombok.Getter;

thinglinks-common/thinglinks-common-redis/src/main/java/com/mqttsnet/thinglinks/common/redis/configure/RedisConfig.java

@@ -6,6 +6,7 @@ import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
 import com.fasterxml.jackson.annotation.JsonAutoDetect;
 import com.fasterxml.jackson.annotation.JsonTypeInfo;
@@ -47,4 +48,16 @@ public class RedisConfig extends CachingConfigurerSupport
         template.afterPropertiesSet();
         return template;
     }
+
+    /**
+     * key过期事件订阅需要
+     * @param redisConnectionFactory
+     * @return
+     */
+    @Bean
+    RedisMessageListenerContainer container(RedisConnectionFactory redisConnectionFactory) {
+        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
+        container.setConnectionFactory(redisConnectionFactory);
+        return container;
+    }
 }

thinglinks-common/thinglinks-common-redis/src/main/java/com/mqttsnet/thinglinks/common/redis/service/RedisService.java

@@ -2,9 +2,12 @@ package com.mqttsnet.thinglinks.common.redis.service;
 
 import java.util.*;
 import java.util.concurrent.TimeUnit;
+
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.connection.DataType;
 import org.springframework.data.redis.core.*;
+import org.springframework.data.redis.core.script.DefaultRedisScript;
 import org.springframework.stereotype.Component;
 
 /**
@@ -14,11 +17,78 @@ import org.springframework.stereotype.Component;
  **/
 @SuppressWarnings(value = { "unchecked", "rawtypes" })
 @Component
+@Slf4j
 public class RedisService
 {
     @Autowired
     public RedisTemplate redisTemplate;
 
+    //分布式锁参数
+    private static final Long RELEASE_SUCCESS = 1L;
+    private static final String RELEASE_LOCK_LUA_SCRIPT = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end";
+
+
+    /**
+     * 判断资源是否已经被锁定
+     *
+     * @param key 锁定的资源标识key
+     * @return 未锁定 true
+     */
+    public boolean checkLock(String key, String requestId, Long time) {
+        //检查key是否存在,获取过期时间
+        Boolean result = redisTemplate.hasKey(key);
+        Long expireTime = redisTemplate.getExpire(key);
+        //当前时间加获取key的时间
+        Long expire = System.currentTimeMillis() + expireTime;
+        //判断该锁是否被获得,锁已经被其他请求获得，直接返回
+        if (result != null && result && expire >= System.currentTimeMillis()) {
+            log.info("该锁已被其他用户获得-key：{}，value：{}", key, requestId);
+            return false;
+        }
+       /* if (result != null && expire < System.currentTimeMillis()) {
+            // 移除过期的锁
+            this.releaseDistributedLock(key, requestId);
+        }*/
+        // (在高并发前提下)在当前请求已经获得锁的前提下，还可能有其他请求尝试去获得锁，此时会导致当前锁的过期时间被延长，由于延长时间在毫秒级，可以忽略。
+        return this.setLock(key, requestId, time);
+    }
+
+
+    /**
+     * 尝试获取分布式锁
+     *
+     * @param lockKey    锁
+     * @param requestId  请求标识
+     * @param expireTime 超期时间
+     * @return 是否获取成功
+     */
+    public boolean setLock(String lockKey, String requestId, Long expireTime) {
+        Boolean result = redisTemplate.opsForValue().setIfAbsent(lockKey, requestId, expireTime, TimeUnit.MILLISECONDS);
+        if (result != null && result) {
+            return true;
+        } else {
+            return false;
+        }
+    }
+
+    /**
+     * 释放分布式锁
+     *
+     * @param lockKey   锁
+     * @param requestId 请求标识
+     * @return 是否释放成功
+     */
+    public boolean releaseDistributedLock(String lockKey, String requestId) {
+        // 指定 lua 脚本，并且指定返回值类型
+        DefaultRedisScript<Long> redisScript = new DefaultRedisScript<>(RELEASE_LOCK_LUA_SCRIPT, Long.class);
+        // 参数一：redisScript，参数二：key列表，参数三：arg（可多个）
+        Long result = (Long) redisTemplate.execute(redisScript, Collections.singletonList(lockKey), requestId);
+        if (RELEASE_SUCCESS.equals(result)) {
+            return true;
+        }
+        return false;
+    }
+
     /**
      * 缓存基本的对象，Integer、String、实体类等
      *
@@ -465,6 +535,20 @@ public class RedisService
         return redisTemplate.opsForValue().setIfAbsent(key, value);
     }
 
+    /**
+     * 只有在 key 不存在时设置 key 的值
+     *
+     * @param key
+     * @param value
+     * @param timeout
+     *            过期时间
+     * @param unit
+     *            时间单位, 天:TimeUnit.DAYS 小时:TimeUnit.HOURS 分钟:TimeUnit.MINUTES
+     *            秒:TimeUnit.SECONDS 毫秒:TimeUnit.MILLISECONDS
+     * @return 之前已经存在返回false,不存在返回true
+     */
+    public boolean setIfAbsent(String key, String value, long timeout, TimeUnit unit) { return redisTemplate.opsForValue().setIfAbsent(key, value, timeout, unit);}
+
     /**
      * 用 value 参数覆写给定 key 所储存的字符串值，从偏移量 offset 开始
      *

thinglinks-gateway/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 服务注册地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置
@@ -35,16 +34,16 @@ spring:
       eager: true
       transport:
         # 控制台地址
-        dashboard: 49.235.122.136:19101
+        dashboard: 127.0.0.1:19101
         port: 19101
       # nacos配置持久化
       datasource:
         ds1:
           nacos:
-            server-addr: 49.235.122.136:8848
+            server-addr: ${spring.cloud.nacos.discovery.server-addr}
             #命名空间
-            namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
-            dataId: sentinel-mqtts-gateway
+            namespace: ${spring.cloud.nacos.discovery.namespace}
+            dataId: sentinel-thinglinks-gateway
             groupId: DEFAULT_GROUP
             data-type: json
             rule-type: flow

thinglinks-modules/thinglinks-modules-broker/pom.xml

@@ -85,6 +85,7 @@
                 </exclusion>
             </exclusions>
         </dependency>
+
         <dependency>
             <groupId>io.projectreactor</groupId>
             <artifactId>reactor-core</artifactId>

thinglinks-modules/thinglinks-modules-broker/src/main/java/com/mqttsnet/thinglinks/broker/service/DeviceActionInterceptor.java

-//package com.mqttsnet.thinglinks.broker.service;
-//
-//import cn.hutool.core.date.LocalDateTimeUtil;
-//import com.mqttsnet.thinglinks.link.api.domain.device.entity.Device;
-//import com.mqttsnet.thinglinks.link.api.domain.device.entity.DeviceAction;
-//import io.github.quickmsg.common.channel.MqttChannel;
-//import io.github.quickmsg.common.config.Configuration;
-//import io.github.quickmsg.common.context.ReceiveContext;
-//import io.github.quickmsg.common.interceptor.Interceptor;
-//import io.github.quickmsg.common.interceptor.Invocation;
-//import io.github.quickmsg.common.message.HeapMqttMessage;
-//import io.github.quickmsg.common.message.SmqttMessage;
-//import io.github.quickmsg.common.rule.DslExecutor;
-//import io.github.quickmsg.common.utils.MessageUtils;
-//import io.netty.handler.codec.mqtt.*;
-//import lombok.extern.slf4j.Slf4j;
-//import com.mqttsnet.thinglinks.link.api.RemoteDeviceActionService;
-//import com.mqttsnet.thinglinks.link.api.RemoteDeviceService;
-//import org.springframework.beans.factory.annotation.Autowired;
-//import org.springframework.stereotype.Component;
-//import org.springframework.stereotype.Service;
-//
-//import javax.annotation.PostConstruct;
-//import java.util.Arrays;
-//import java.util.List;
-//
-///**
-// * @Description: Mqtt 设备动作拦截处理
-// * @Author: ShiHuan Sun
-// * @E-mail: 13733918655@163.com
-// * @Website: http://thinglinks.mqttsnet.com
-// * @CreateDate: 2021/11/16$ 10:33$
-// * @UpdateUser: ShiHuan Sun
-// * @UpdateDate: 2021/11/16$ 10:33$
-// * @UpdateRemark: 修改内容
-// * @Version: 1.0
-// */
-//@Service
-//@Slf4j
-//@Component
-//public class DeviceActionInterceptor implements Interceptor {
-//
-//    private static DeviceActionInterceptor DeviceActionInterceptor;
-//
-//    @Autowired
-//    private RemoteDeviceService deviceService;
-//
-//    @Autowired
-//    private RemoteDeviceActionService deviceActionService;
-//
-//
-//    @PostConstruct
-//    public void init() {
-//        DeviceActionInterceptor = this;
-//        DeviceActionInterceptor.deviceService = this.deviceService;
-//        DeviceActionInterceptor.deviceActionService = this.deviceActionService;
-//    }
-//
-//    /**
-//     * 拦截目标参数
-//     *
-//     * @param invocation {@link Invocation}
-//     * @return Object
-//     */
-//    @Override
-//    public Object intercept(Invocation invocation) {
-//        try {
-//            MqttChannel mqttChannel = (MqttChannel) invocation.getArgs()[0];
-//            SmqttMessage<MqttMessage> smqttMessage = (SmqttMessage<MqttMessage>) invocation.getArgs()[1];
-//            ReceiveContext<Configuration> mqttReceiveContext = (ReceiveContext<Configuration>) invocation.getArgs()[2];
-//            DslExecutor dslExecutor = mqttReceiveContext.getDslExecutor();
-//            MqttMessage message = smqttMessage.getMessage();
-//            //TODO MQTT动作数据处理
-//            List<MqttMessageType> mqttMessageType = Arrays.asList(MqttMessageType.PUBLISH, MqttMessageType.DISCONNECT, MqttMessageType.PINGRESP, MqttMessageType.SUBSCRIBE, MqttMessageType.UNSUBSCRIBE);
-//            if (!smqttMessage.getIsCluster() && mqttMessageType.contains(message.fixedHeader().messageType())) {
-//                MqttPublishMessage publishMessage = (MqttPublishMessage) message;
-//                HeapMqttMessage heapMqttMessage = this.clusterMessage(publishMessage, mqttChannel, smqttMessage.getTimestamp());
-//                DeviceAction deviceAction = new DeviceAction();
-//                deviceAction.setDeviceIdentification(mqttChannel.getClientIdentifier());
-//                deviceAction.setActionType(message.fixedHeader().messageType().toString());
-//                deviceAction.setStatus(message.decoderResult().toString());
-//                deviceAction.setMessage(heapMqttMessage.getTopic());
-//                deviceAction.setCreateTime(LocalDateTimeUtil.now());
-//                DeviceActionInterceptor.deviceActionService.add(deviceAction);
-//                Device device = new Device();
-//                device.setConnectStatus((mqttChannel.getStatus().toString());
-//                device.setDeviceIdentification(mqttChannel.getClientIdentifier());
-//                DeviceActionInterceptor.deviceService.updateConnectStatusByClientId(device);
-//            }
-//            // 拦截业务
-//            return invocation.proceed(); // 放行
-//        } catch (Exception e) {
-//            e.printStackTrace();
-//        }
-//        return null;
-//    }
-//
-//    /**
-//     * 构建消息体
-//     *
-//     * @param message   {@link MqttPublishMessage}
-//     * @param timestamp
-//     * @return {@link HeapMqttMessage}
-//     */
-//    private HeapMqttMessage clusterMessage(MqttPublishMessage message, MqttChannel channel, long timestamp) {
-//        MqttPublishVariableHeader header = message.variableHeader();
-//        MqttFixedHeader fixedHeader = message.fixedHeader();
-//        return HeapMqttMessage.builder()
-//                .timestamp(timestamp)
-//                .clientIdentifier(channel.getClientIdentifier())
-//                .message(MessageUtils.copyReleaseByteBuf(message.payload()))
-//                .topic(header.topicName())
-//                .retain(fixedHeader.isRetain())
-//                .qos(fixedHeader.qosLevel().value())
-//                .build();
-//    }
-//
-//    /**
-//     * 排序
-//     * 值越大权重越高
-//     *
-//     * @return 排序
-//     */
-//    @Override
-//    public int sort() {
-//        return 0;
-//    }
-//}
+package com.mqttsnet.thinglinks.broker.service;
+
+import com.mqttsnet.thinglinks.common.core.constant.Constants;
+import com.mqttsnet.thinglinks.common.core.utils.DateUtils;
+import com.mqttsnet.thinglinks.common.redis.service.RedisService;
+import com.mqttsnet.thinglinks.link.api.RemoteDeviceActionService;
+import com.mqttsnet.thinglinks.link.api.RemoteDeviceService;
+import io.github.quickmsg.common.channel.MqttChannel;
+import io.github.quickmsg.common.config.Configuration;
+import io.github.quickmsg.common.context.ReceiveContext;
+import io.github.quickmsg.common.interceptor.Interceptor;
+import io.github.quickmsg.common.interceptor.Invocation;
+import io.github.quickmsg.common.message.HeapMqttMessage;
+import io.github.quickmsg.common.message.SmqttMessage;
+import io.github.quickmsg.common.rule.DslExecutor;
+import io.github.quickmsg.common.utils.MessageUtils;
+import io.netty.handler.codec.mqtt.*;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description: Mqtt 设备动作拦截处理
+ * @Author: ShiHuan Sun
+ * @E-mail: 13733918655@163.com
+ * @Website: http://thinglinks.mqttsnet.com
+ * @CreateDate: 2021/11/16$ 10:33$
+ * @UpdateUser: ShiHuan Sun
+ * @UpdateDate: 2021/11/16$ 10:33$
+ * @UpdateRemark: 修改内容
+ * @Version: 1.0
+ */
+@Service
+@Slf4j
+@Component
+public class DeviceActionInterceptor implements Interceptor {
+
+    private static DeviceActionInterceptor DeviceActionInterceptor;
+
+    @Autowired
+    private RemoteDeviceService deviceService;
+
+    @Autowired
+    private RemoteDeviceActionService deviceActionService;
+
+    @Autowired
+    private RedisService redisService;
+
+
+    @PostConstruct
+    public void init() {
+        DeviceActionInterceptor = this;
+        DeviceActionInterceptor.deviceService = this.deviceService;
+        DeviceActionInterceptor.deviceActionService = this.deviceActionService;
+        DeviceActionInterceptor.redisService = this.redisService;
+    }
+
+    /**
+     * 拦截目标参数
+     *
+     * @param invocation {@link Invocation}
+     * @return Object
+     */
+    @Override
+    public Object intercept(Invocation invocation) {
+        try {
+            MqttChannel mqttChannel = (MqttChannel) invocation.getArgs()[0];
+            SmqttMessage<MqttMessage> smqttMessage = (SmqttMessage<MqttMessage>) invocation.getArgs()[1];
+            ReceiveContext<Configuration> mqttReceiveContext = (ReceiveContext<Configuration>) invocation.getArgs()[2];
+            DslExecutor dslExecutor = mqttReceiveContext.getDslExecutor();
+            MqttMessage message = smqttMessage.getMessage();
+            //TODO MQTT设备心跳处理
+            List<MqttMessageType> mqttMessageType = Collections.singletonList(MqttMessageType.PINGREQ);
+            if (!smqttMessage.getIsCluster() && mqttMessageType.contains(message.fixedHeader().messageType())) {
+                DeviceActionInterceptor.redisService.expire(Constants.DEVICE_RECORD_KEY+mqttChannel.getClientIdentifier(),300L+ Long.parseLong(DateUtils.getRandom(1)), TimeUnit.SECONDS);
+            }
+            return invocation.proceed(); // 放行
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return null;
+    }
+
+    /**
+     * 构建消息体
+     *
+     * @param message   {@link MqttPublishMessage}
+     * @param timestamp
+     * @return {@link HeapMqttMessage}
+     */
+    private HeapMqttMessage clusterMessage(MqttPublishMessage message, MqttChannel channel, long timestamp) {
+        MqttPublishVariableHeader header = message.variableHeader();
+        MqttFixedHeader fixedHeader = message.fixedHeader();
+        return HeapMqttMessage.builder()
+                .timestamp(timestamp)
+                .clientIdentifier(channel.getClientIdentifier())
+                .message(MessageUtils.copyReleaseByteBuf(message.payload()))
+                .topic(header.topicName())
+                .retain(fixedHeader.isRetain())
+                .qos(fixedHeader.qosLevel().value())
+                .build();
+    }
+
+    /**
+     * 排序
+     * 值越大权重越高
+     *
+     * @return 排序
+     */
+    @Override
+    public int sort() {
+        return 0;
+    }
+}

thinglinks-modules/thinglinks-modules-broker/src/main/java/com/mqttsnet/thinglinks/broker/service/PasswordAuthenticationImpl.java

 package com.mqttsnet.thinglinks.broker.service;
 
 
+import com.mqttsnet.thinglinks.common.core.constant.Constants;
+import com.mqttsnet.thinglinks.common.core.enums.DeviceConnectStatus;
+import com.mqttsnet.thinglinks.common.core.utils.DateUtils;
+import com.mqttsnet.thinglinks.common.redis.service.RedisService;
 import com.mqttsnet.thinglinks.link.api.domain.device.entity.Device;
 import io.github.quickmsg.common.auth.PasswordAuthentication;
 import lombok.extern.slf4j.Slf4j;
@@ -11,6 +15,7 @@ import org.springframework.stereotype.Service;
 
 import javax.annotation.PostConstruct;
 import java.util.Optional;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @Description: MQTT 连接鉴权
@@ -31,12 +36,15 @@ public class PasswordAuthenticationImpl implements PasswordAuthentication {
 
     @Autowired
     private RemoteDeviceService deviceService;
+    @Autowired
+    private RedisService redisService;
 
 
     @PostConstruct
     public void init() {
         PasswordAuthenticationImpl = this;
         PasswordAuthenticationImpl.deviceService = this.deviceService;
+        PasswordAuthenticationImpl.redisService = this.redisService;
     }
 
     /**
@@ -51,9 +59,11 @@ public class PasswordAuthenticationImpl implements PasswordAuthentication {
     public boolean auth(String userName, byte[] passwordInBytes, String clientIdentifier) {
         Device mqttsDevice = PasswordAuthenticationImpl.deviceService.findOneByClientIdAndUserNameAndPasswordAndDeviceStatusAndProtocolType(clientIdentifier, userName, new String(passwordInBytes), "ENABLE", "MQTT").getData();
         if (Optional.ofNullable(mqttsDevice).isPresent()) {
+            //缓存设备信息
+            PasswordAuthenticationImpl.redisService.setCacheObject(Constants.DEVICE_RECORD_KEY+mqttsDevice.getClientId(),mqttsDevice,300L+ Long.parseLong(DateUtils.getRandom(1)), TimeUnit.SECONDS);
             //更改设备在线状态为在线
             Device device = new Device();
-            device.setConnectStatus("ONLINE");
+            device.setConnectStatus(DeviceConnectStatus.ONLINE.getValue());
             device.setDeviceIdentification(clientIdentifier);
             PasswordAuthenticationImpl.deviceService.updateConnectStatusByClientId(device);
             return true;

thinglinks-modules/thinglinks-modules-broker/src/main/resources/META-INF/services/io.github.quickmsg.common.interceptor.Interceptor

+com.mqttsnet.thinglinks.broker.service.DeviceActionInterceptor
\ No newline at end of file

thinglinks-modules/thinglinks-modules-broker/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-modules/thinglinks-modules-file/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-modules/thinglinks-modules-gen/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-modules/thinglinks-modules-job/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/aop/NoRepeatSubmitAop.java

 package com.mqttsnet.thinglinks.link.common.aop;
 
 import com.mqttsnet.thinglinks.common.core.annotation.NoRepeatSubmit;
+import com.mqttsnet.thinglinks.common.core.constant.Constants;
 import com.mqttsnet.thinglinks.common.core.utils.SecurityUtils;
 import com.mqttsnet.thinglinks.common.core.web.domain.AjaxResult;
 import com.mqttsnet.thinglinks.common.redis.service.RedisService;
@@ -38,9 +39,9 @@ public class NoRepeatSubmitAop {
             log.info("请求地址：{}", request.getServletPath());
             String key = SecurityUtils.getToken() + "-" + request.getServletPath();
             log.info("newToken:{}", key);
-            if (!redisService.hasKey(key)) {// 如果缓存中有这个url视为重复提交
+            if (!redisService.hasKey(Constants.RESUBMIT_URL_KEY+key)) {// 如果缓存中有这个url视为重复提交
                 Object o = pjp.proceed();//当使用环绕通知时，这个方法必须调用，否则拦截到的方法就不会再执行了
-                redisService.setCacheObject(key, o, 2L, TimeUnit.SECONDS);
+                redisService.setCacheObject(Constants.RESUBMIT_URL_KEY+key, o, 2L, TimeUnit.SECONDS);
                 return o;
             } else {
                 log.error("请勿重复提交");

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/common/listener/RedisKeyExpirationListener.java

+package com.mqttsnet.thinglinks.link.common.listener;
+
+import com.mqttsnet.thinglinks.common.core.constant.Constants;
+import com.mqttsnet.thinglinks.common.core.text.UUID;
+import com.mqttsnet.thinglinks.common.redis.service.RedisService;
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.connection.Message;
+import org.springframework.data.redis.listener.KeyExpirationEventMessageListener;
+import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @Description: Redis Key失效事件监听
+ * @Author: ShiHuan SUN
+ * @E-mail: 13733918655@163.com
+ * @Website: http://thinglinks.mqttsnet.com
+ * @CreateDate: 2022/3/1$ 15:28$
+ * @UpdateUser: ShiHuan SUN
+ * @UpdateDate: 2022/3/1$ 15:28$
+ * @UpdateRemark: 修改内容
+ * @Version: V1.0
+ */
+@Component
+@Slf4j
+public class RedisKeyExpirationListener extends KeyExpirationEventMessageListener {
+
+    @Autowired
+    private RedisService redisService;
+
+    public RedisKeyExpirationListener(RedisMessageListenerContainer listenerContainer) {
+        super(listenerContainer);
+    }
+
+    @SneakyThrows
+    @Override
+    public void onMessage(Message message, byte[] pattern) {
+        String expiredKey = Constants.SET_NX+message.toString();
+        // 避免多个服务监听情况下重复消费
+        boolean resultLock = false;
+        String uuid = UUID.getUUID();
+        try {
+            resultLock = redisService.checkLock(expiredKey, uuid, 10L);
+            log.info("获取分布式锁返回值：{}", resultLock);
+            if (resultLock) {
+                log.info("获取分布式锁成功-key：{}，value：{}", expiredKey, uuid);
+                if (expiredKey.contains(Constants.DEVICE_RECORD_KEY)){
+                    log.info("设备信息缓存失效{}",expiredKey);
+                }
+            }else {
+                log.info("获取分布式锁失败-key：{}，value：{}", expiredKey, uuid);
+            }
+        }catch (Exception e){
+            log.error(e.getMessage());
+        }finally {
+            if (resultLock) {
+                //释放锁
+                boolean result = redisService.releaseDistributedLock(expiredKey, uuid);
+                if (result) {
+                    log.info("释放分布式锁成功-key：{}，value：{}", expiredKey, uuid);
+                } else {
+                    log.info("释放分布式锁失败-key：{}，value：{}", expiredKey, uuid);
+                }
+            }
+        }
+    }
+}

thinglinks-modules/thinglinks-modules-link/src/main/java/com/mqttsnet/thinglinks/link/service/device/impl/DeviceServiceImpl.java

 package com.mqttsnet.thinglinks.link.service.device.impl;
 
+import com.mqttsnet.thinglinks.common.core.enums.DeviceConnectStatus;
 import com.mqttsnet.thinglinks.common.core.utils.DateUtils;
 import com.mqttsnet.thinglinks.common.core.utils.StringUtils;
 import com.mqttsnet.thinglinks.common.security.service.TokenService;
@@ -180,6 +181,7 @@ public class DeviceServiceImpl implements DeviceService {
         if(StringUtils.isNotNull(oneByClientIdAndDeviceIdentification)){
             return 0;
         }
+        device.setConnectStatus(DeviceConnectStatus.INIT.getValue());
         LoginUser loginUser = tokenService.getLoginUser();
         SysUser sysUser = loginUser.getSysUser();
         device.setCreateBy(sysUser.getUserName());

thinglinks-modules/thinglinks-modules-link/src/main/resources/bootstrap.yml

@@ -11,22 +11,22 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置
         shared-configs:
           - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
+

thinglinks-modules/thinglinks-modules-system/src/main/resources/bootstrap.yml

@@ -11,20 +11,19 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置

thinglinks-modules/thinglinks-modules-tdengine/src/main/resources/bootstrap.yml

@@ -11,22 +11,21 @@ spring:
     # 环境配置
     active: dev
   main:
-    allow-circular-references: true
     allow-bean-definition-overriding: true
   cloud:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: ${spring.cloud.nacos.discovery.server-addr}
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: ${spring.cloud.nacos.discovery.namespace}
         # 配置文件格式
         file-extension: yml
         # 共享配置
         shared-configs:
-          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
+          - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
\ No newline at end of file

thinglinks-visual/thinglinks-visual-collection/pom.xml

@@ -78,11 +78,6 @@
             <version>5.10.0</version>
         </dependency>
 
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-        </dependency>
-
         <!-- thinglinks-api-monitor -->
         <dependency>
             <groupId>com.mqttsnet</groupId>

thinglinks-visual/thinglinks-visual-collection/src/main/resources/bootstrap.yml

@@ -17,12 +17,12 @@ spring:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
         namespace: 05545134-ed4b-4647-89ff-f83bed4db8f2
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
         namespace: 05545134-ed4b-4647-89ff-f83bed4db8f2
         # 配置文件格式

thinglinks-visual/thinglinks-visual-monitor/src/main/resources/bootstrap.yml

@@ -17,14 +17,14 @@ spring:
     nacos:
       discovery:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
       config:
         # 配置中心地址
-        server-addr: 49.235.122.136:8848
+        server-addr: 127.0.0.1:8848
         #命名空间
-        namespace: 1e1aff6c-da73-43e2-9e5f-8e0b890189d9
+        namespace: 8ea40c2e-64ba-4430-9bd8-a25336b2b45a
         # 配置文件格式
         file-extension: yml
         # 关联配置