mqtt设备上下线监听

device-api-service/src/main/java/com/xy/consumer/ConnectedConsumer.java

@@ -0,0 +1,46 @@
+package com.xy.consumer;
+
+import cn.hutool.json.JSONObject;
+import cn.hutool.json.JSONUtil;
+import com.xy.dto.DeviceStatusDto;
+import com.xy.service.DeviceStatusService;
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 设备上线监听消费者
+ * </p>
+ *
+ * @author lijin
+ * @since 2022-12-27
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class ConnectedConsumer implements MqttConsumer {
+
+    private DeviceStatusService deviceStatusService;
+
+    @Override
+    public boolean message(String topic, String payload) {
+        log.info("设备上线监听:{}", payload);
+        return exc(payload, true);
+    }
+
+    public boolean exc(String payload, boolean netState) {
+        JSONObject jsonObject = JSONUtil.parseObj(payload);
+        String clientid = jsonObject.getStr("clientid");
+        Long deviceId;
+        try {
+            deviceId = Long.valueOf(clientid);
+        } catch (Exception e) {
+            return true;
+        }
+        DeviceStatusDto.Up up = new DeviceStatusDto.Up().setDeviceId(deviceId);
+        up.setNetState(netState);
+        int code = deviceStatusService.up(up).getCode();
+        return code == 200;
+    }
+}

device-api-service/src/main/java/com/xy/consumer/DisconnectedConsumer.java

@@ -0,0 +1,27 @@
+package com.xy.consumer;
+
+import lombok.AllArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 设备下线监听消费者
+ * </p>
+ *
+ * @author lijin
+ * @since 2022-12-27
+ */
+@Slf4j
+@Service
+@AllArgsConstructor
+public class DisconnectedConsumer implements MqttConsumer {
+
+    private ConnectedConsumer connectedConsumer;
+
+    @Override
+    public boolean message(String topic, String payload) {
+        log.info("设备下线监听:{}", payload);
+        return connectedConsumer.exc(payload, false);
+    }
+}

device-api-service/src/main/java/com/xy/entity/DeviceStatus.java

@@ -47,7 +47,7 @@ public class DeviceStatus extends BaseEntity3<DeviceStatus> {
     /**
      * 联网状态
      */
-    private Integer netState;
+    private Boolean netState;
 
     /**
      * 联网类型

device-api-service/src/main/java/com/xy/service/impl/DeviceStatusServiceImpl.java

@@ -27,6 +27,10 @@ public class DeviceStatusServiceImpl extends ServiceImpl<DeviceStatusMapper, Dev
     @Override
     @ApiOperation("上报信息")
     public R up(DeviceStatusDto.Up up) {
+        DeviceStatus deviceStatusInfo = getById(up.getDeviceId());
+        if (deviceStatusInfo == null) {
+            return R.ok();
+        }
         DeviceStatus deviceStatus = copy(DeviceStatus.class, up);
         updateById(deviceStatus);
         return R.ok();

device-api/src/main/java/com/xy/dto/DeviceStatusDto.java

@@ -41,7 +41,7 @@ public class DeviceStatusDto {
         private String bloothOpen;
 
         @ApiModelProperty("联网状态")
-        private Integer netState;
+        private Boolean netState;
 
         @ApiModelProperty("联网类型")
         private String netType;

device-start/src/main/resources/bootstrap-uat.yml

@@ -2,6 +2,8 @@ spring:
   mqtt:
     topics:
       - {topic: "device-log", handler: com.xy.consumer.LogConsumer}
+      - {topic: "device-disconnected", handler: com.xy.consumer.DisconnectedConsumer}
+      - {topic: "device-connected", handler: com.xy.consumer.ConnectedConsumer}
 
 #微服务相关配置
 cloud:

device-start/src/main/resources/bootstrap.yml

@@ -5,6 +5,8 @@ spring:
     is-retry: true
     topics:
       - {topic: "dev-device-log", handler: com.xy.consumer.LogConsumer}
+      - {topic: "device-disconnected", handler: com.xy.consumer.DisconnectedConsumer}
+      - {topic: "device-connected", handler: com.xy.consumer.ConnectedConsumer}
 
 server:
   port: 9051