添加联网记录

device-api-service/src/main/java/com/xy/config/DeviceThreadPoolConfig.java

@@ -14,6 +14,8 @@ public class DeviceThreadPoolConfig {
 
     public static final String DEVICE_DATA_POLL = "deviceDataPoll";
 
+    public static final String DEVICE_NETWORK_POLL = "deviceNetWorkPoll";
+
     /**
      * 公用线程池
      */
@@ -49,4 +51,22 @@ public class DeviceThreadPoolConfig {
                 .builder();
     }
 
+    /**
+     * 上下线mqtt消费线程池
+     */
+    @DynamicTp
+    @Bean(DEVICE_NETWORK_POLL)
+    public ThreadPoolTaskExecutor deviceNetWorkPoll() {
+        int coreSize = 5;
+        return ThreadPoolUtils.newPoll()
+                .name(DEVICE_NETWORK_POLL)
+                .coreSize(coreSize)
+                .maxSize(coreSize * 2)
+                .keepAlive(60)
+                .queueSize(coreSize * 10)
+                .rejected((r, executor) -> {
+                    throw new CommRuntimeException("统计消费者线程池已满");
+                })
+                .builder();
+    }
 }

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

@@ -2,9 +2,14 @@ package com.xy.consumer.connected;
 
 import cn.hutool.json.JSONObject;
 import cn.hutool.json.JSONUtil;
+import com.xy.config.DeviceThreadPoolConfig;
 import com.xy.consumer.MqttConsumer;
+import com.xy.dto.DeviceNetRecordDto;
 import com.xy.dto.DeviceStatusDto;
+import com.xy.service.DeviceNetRecordServiceImpl;
 import com.xy.service.DeviceStatusServiceImpl;
+import com.xy.utils.ThreadPoolUtils;
+import com.xy.utils.enums.DictSonEnum;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -24,10 +29,11 @@ public class ConnectedConsumer implements MqttConsumer {
 
     private DeviceStatusServiceImpl deviceStatusService;
 
+    private DeviceNetRecordServiceImpl deviceNetRecordService;
+
     @Override
     public boolean message(String topic, String payload) {
-        log.info("设备上线监听:{}", payload);
-        return exc(payload, 1);
+        return exc(payload, DictSonEnum.DEVICE_ONLINE_STATUS_1.getKeyInt());
     }
 
     public boolean exc(String payload, int netState) {
@@ -39,9 +45,20 @@ public class ConnectedConsumer implements MqttConsumer {
         } 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;
+        String msg = netState == DictSonEnum.DEVICE_ONLINE_STATUS_1.getKeyInt() ? "上线" : "离线";
+        log.info("设备" + msg + "监听:{}", payload);
+        ThreadPoolUtils.excPoll(DeviceThreadPoolConfig.DEVICE_NETWORK_POLL, 1)
+                .execute(() -> {
+                    //上报状态
+                    DeviceStatusDto.Up up = new DeviceStatusDto.Up().setDeviceId(deviceId);
+                    up.setNetState(netState);
+                    deviceStatusService.up(up).getCode();
+                    //添加联网记录
+                    DeviceNetRecordDto.Save save = new DeviceNetRecordDto.Save()
+                            .setDeviceId(deviceId);
+                    save.setNetStatus(netState);
+                    deviceNetRecordService.save(new DeviceNetRecordDto.Save());
+                });
+        return true;
     }
 }

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

@@ -2,6 +2,7 @@ package com.xy.consumer.disconnect;
 
 import com.xy.consumer.MqttConsumer;
 import com.xy.consumer.connected.ConnectedConsumer;
+import com.xy.utils.enums.DictSonEnum;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
@@ -23,7 +24,6 @@ public class DisconnectedConsumer implements MqttConsumer {
 
     @Override
     public boolean message(String topic, String payload) {
-        log.info("设备下线监听:{}", payload);
-        return connectedConsumer.exc(payload, 2);
+        return connectedConsumer.exc(payload, DictSonEnum.DEVICE_ONLINE_STATUS_2.getKeyInt());
     }
 }

device-api-service/src/main/java/com/xy/utils/enums/DictSonEnum.java

@@ -44,10 +44,17 @@ public enum DictSonEnum {
 
     DEVICE_BUSY_STATUS_2("2", "设备-运营状态-已停运"),
 
+    DEVICE_ONLINE_STATUS_1("1", "设备-联网状态-在线"),
+
+    DEVICE_ONLINE_STATUS_2("2", "设备-联网状态-离线"),
+
     ;
 
     private String key;
 
     private String msg;
 
+    public int getKeyInt() {
+        return Integer.valueOf(this.key);
+    }
 }