#init

maibu-netty-server/src/main/java/com/maibu/service/DeviceThreadService.java

@@ -0,0 +1,166 @@
+package com.fastbee.service;
+
+
+import cn.hutool.core.io.IoUtil;
+import com.alibaba.fastjson2.JSON;
+import com.fastbee.common.CommandConstant;
+import com.fastbee.common.NettyCacheKey;
+import com.fastbee.common.core.redis.RedisCache;
+import com.fastbee.dto.DeviceErrorPushDTO;
+import com.fastbee.dto.DeviceErrorPushDetail;
+import com.fastbee.dto.NettyDevice;
+import com.fastbee.iot.domain.DeviceRunningStatusHistory;
+import com.fastbee.entity.ErrorIdentificationStandard;
+import com.fastbee.enums.CompareEnum;
+import com.fastbee.enums.RespondCode;
+import com.fastbee.iot.domain.DeviceStatusRecordDTO;
+import com.fastbee.manager.DeviceSessionManager;
+import com.fastbee.mapper.ErrorIdentificationStandardMapper;
+import com.fastbee.memory.GlobalMemory;
+import com.fastbee.netty.handler.DeviceConnectHandler;
+import com.fastbee.utils.CommandUtils;
+import com.fastbee.utils.CompareUtils;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.core.io.Resource;
+import org.springframework.core.io.ResourceLoader;
+import org.springframework.stereotype.Service;
+
+import java.io.IOException;
+import java.lang.reflect.Field;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+@Slf4j
+@Service
+public class DeviceThreadService {
+
+    @Autowired
+    private ResourceLoader resourceLoader;
+
+    @Autowired
+    private GlobalMemory globalMemory;
+
+    @Autowired
+    private RedisCache redisCache;
+
+    @Autowired
+    private DeviceSessionManager sessionManager;
+
+    @Autowired
+    private ErrorIdentificationStandardMapper standardMapper;
+
+
+    public void deviceErrorMonitor() throws InterruptedException, IOException {
+        initStandard();
+
+        ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
+        executor.scheduleWithFixedDelay(() -> {
+            try {
+                String key = NettyCacheKey.deviceRunningStatusKey;
+                Collection<String> keys = redisCache.getListKeyByPrefix(key);
+                if (!CollectionUtils.isEmpty(keys)) {
+                    keys.forEach(k -> {
+                        DeviceStatusRecordDTO recordDTO = redisCache.getCacheObject(k);
+                        if (recordDTO != null && !CollectionUtils.isEmpty(recordDTO.getHistories())) {
+                            recordDTO.getHistories().sort(Comparator.comparing(DeviceRunningStatusHistory::getCreateTime).reversed());
+                            DeviceRunningStatusHistory history = recordDTO.getHistories().get(0);
+                            List<DeviceErrorPushDetail> compared = compareErrorStandard(history, globalMemory.standard);
+
+                            String[] parts = k.split(":");
+                            String deviceId = parts[1];
+
+                            DeviceErrorPushDTO pushDTO = new DeviceErrorPushDTO();
+                            pushDTO.setDeviceId(deviceId);
+                            pushDTO.setTime(System.currentTimeMillis());
+                            pushDTO.setDetails(compared);
+                            pushDTO.setEvent(RespondCode.device_error_push);
+                            //todo push 推送到前端
+                            pushErrorMessage(deviceId, JSON.toJSONString(pushDTO));
+                        }
+                    });
+                }
+            } catch (Exception e) {
+                log.error("执行监测错误线程错误：{}", e.getMessage());
+            }
+        }, 0, 2, TimeUnit.MINUTES);
+    }
+
+    public void initStandard() throws IOException {
+        if (globalMemory.standard == null) {
+            List<ErrorIdentificationStandard> standards = standardMapper.selectList();
+            if (CollectionUtils.isEmpty(standards)) {
+                // 读取resource目录下的device-config.json文件
+                Resource resource = resourceLoader.getResource("classpath:errorStandard.json");
+                // 使用fastjson2转换为DeviceConfig对象
+                String jsonContent = IoUtil.readUtf8(resource.getInputStream());
+                globalMemory.standard = JSON.parseArray(jsonContent, ErrorIdentificationStandard.class);
+            } else {
+                globalMemory.standard = standards;
+            }
+        }
+    }
+
+    public void pushErrorMessage(String deviceId, String dto) {
+        NettyDevice device = sessionManager.getDevice(deviceId);
+        if (device != null) {
+            ByteBuf buf = Unpooled.buffer();
+            CommandUtils.buildCommand(buf, dto, CommandConstant.interaction);
+            if (device.getChannel() != null && device.getChannel().isActive()) {
+                device.getChannel().writeAndFlush(buf);
+            }
+        }
+    }
+
+
+    public List<DeviceErrorPushDetail> compareErrorStandard(DeviceRunningStatusHistory history, List<ErrorIdentificationStandard> standards) {
+        if (history == null || CollectionUtils.isEmpty(standards)) return null;
+        List<DeviceErrorPushDetail> list = new ArrayList<>();
+        standards.forEach(standard -> {
+            String fieldName = standard.getField();
+            String compareValues = standard.getCompareValues();
+            CompareEnum compareType = standard.getCompareType();
+            String targetValue = knownClassFieldFinding(fieldName, history);
+            if (!StringUtils.isEmpty(compareValues) && !StringUtils.isEmpty(targetValue)) {
+                boolean result = CompareUtils.compare(compareType, compareValues, targetValue);
+                if (!result) {
+                    DeviceErrorPushDetail detail = new DeviceErrorPushDetail();
+                    detail.setErrorName(standard.getErrorName());
+                    detail.setDescription(standard.getErrorDescription());
+                    detail.setValue(targetValue);
+                    detail.setRange(compareValues);
+                    list.add(detail);
+                }
+            }
+        });
+        return list;
+    }
+
+
+    private static String knownClassFieldFinding(String fieldName, DeviceRunningStatusHistory history) {
+        try {
+            // 1. 获取类的Class对象
+            Class<?> historyClass = DeviceRunningStatusHistory.class;
+            Field field = historyClass.getDeclaredField(fieldName);
+            field.setAccessible(true);
+            Object value = field.get(history);
+            return (String) value;
+        } catch (NoSuchFieldException | IllegalAccessException e) {
+            System.out.println("获取字段失败: " + e.getMessage());
+            log.error("获取属性值失败:{}", e.getMessage());
+        }
+        return null;
+    }
+
+}