json解析,增加根据规则配置的数据类型解析

iot-common/iot-common-core/src/main/java/com/shkpr/iot/common/core/enums/ValueTypeEnum.java

@@ -28,8 +28,7 @@ public enum ValueTypeEnum {
      */
     ASCII,
     /**
-     * 布尔(一般用不上)
+     * 布尔
      */
-    @Deprecated
     BOOLEAN,
 }

iot-common/iot-common-core/src/main/java/com/shkpr/iot/common/core/util/JsonUtil.java

@@ -6,7 +6,9 @@ import com.shkpr.iot.common.core.constants.TemplateMetadata;
 import com.shkpr.iot.common.core.domain.po.DataValue;
 import com.shkpr.iot.common.core.domain.po.PointRule;
 import com.shkpr.iot.common.core.domain.po.Template;
+import com.shkpr.iot.common.core.enums.ValueTypeEnum;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.math.NumberUtils;
 
 import java.io.IOException;
 import java.time.LocalDateTime;
@@ -90,11 +92,11 @@ public class JsonUtil {
                 PointRule rule = rules.stream().filter(it -> {
                     String[] split = it.getFlagIndex().split(",");
                     return split.length == 2 &&
-                            Objects.equals(objectMapper.convertValue(node.get(split[0]),String.class), split[1]);
+                            Objects.equals(objectMapper.convertValue(node.get(split[0]), String.class), split[1]);
                 }).findFirst().orElse(null);
                 if (rule == null) continue;
                 //获取值
-                Object value = objectMapper.convertValue(node.get(rule.getValueIndex()), Object.class);
+                Object value = getValue(node.get(rule.getValueIndex()), rule.getValueType());
                 //存入值对象
                 dataValues.add(new DataValue(equipmentCode, rule.getKey(), value, dateTime));
             }
@@ -106,4 +108,38 @@ public class JsonUtil {
         }
         return Collections.emptyList();
     }
+
+    /**
+     * 获取值
+     * @param node 节点
+     * @param type 类型
+     * @return 值
+     */
+    private static Object getValue(JsonNode node, ValueTypeEnum type) {
+        if (node == null || type == null || node.isNull()) return null;
+        //按字符串解析json
+        String s = String.valueOf(objectMapper.convertValue(node, Object.class)).trim();
+        //根据类型解析值
+        Object value;
+        switch (type) {
+            case BOOLEAN:
+                if (NumberUtils.isParsable(s)) value = Integer.parseInt(s) == 1;
+                else value = Boolean.valueOf(s);
+                break;
+            case UNSIGNED_INT:
+            case SIGNED_INT:
+                value = Integer.parseInt(s);
+                break;
+            case SINGLE_FLOAT:
+                value = Float.parseFloat(s);
+                break;
+            case DOUBLE_FLOAT:
+                value = Double.parseDouble(s);
+                break;
+            case ASCII:
+            default:
+                value = s;
+        }
+        return value;
+    }
 }

iot-driver/iot-driver-mqtt/src/main/java/com/shkpr/iot/driver/mqtt/service/MqttService.java

@@ -175,7 +175,7 @@ public class MqttService implements DriverService {
                     try {
                         mqttClient.subscribe(destination.getTopic(), 2);
                     } catch (MqttException e) {
-                        throw new RuntimeException(e);
+                        log.error("MQTT客户端关闭失败,通讯标识:{},{}", destination.getHost(), destination.getPort(), e);
                     }
                     //回调活跃
                     mqttInboundHandler.channelActive();
@@ -190,7 +190,7 @@ public class MqttService implements DriverService {
                         if (CLIENT_CACHE.containsKey(destination) && CLIENT_CACHE.get(destination) != null)
                             CLIENT_CACHE.remove(destination).close(true);
                     } catch (MqttException e) {
-                        log.error("MQTT客户端关闭失败,通讯标识:{},{}", destination.getHost(), destination.getPort(), exception);
+                        log.error("MQTT客户端关闭失败,通讯标识:{},{}", destination.getHost(), destination.getPort(), e);
                     }
                 }
             });

iot-server/iot-server-data/src/main/java/com/shkpr/iot/server/data/service/impl/DataServiceImpl.java

@@ -54,12 +54,24 @@ public class DataServiceImpl implements DataService {
                     .atZone(ZoneId.systemDefault())
                     .toInstant()
                     .toEpochMilli();
-            //构建点
-            Point point = Point.measurement("WaterQuality")
+            //从基本信息创建构建器
+            Point.Builder builder = Point.measurement("WaterQuality")
                     .tag(InfluxdbMetadata.FLAG_EQUIPMENT_CODE, data.getEquipmentCode())
-                    .addField(dataValue.getKey(), (String) dataValue.getValue())
-                    .time(timestamp, TimeUnit.MILLISECONDS)
-                    .build();
+                    .time(timestamp, TimeUnit.MILLISECONDS);
+            //根据类型存入字段
+            Object value = dataValue.getValue();
+            if (value instanceof Boolean) builder.addField(dataValue.getKey(), (Boolean) value);
+            else if (value instanceof Long) builder.addField(dataValue.getKey(), (Long) value);
+            else if (value instanceof Double) builder.addField(dataValue.getKey(), (Double) value);
+            else if (value instanceof Integer) builder.addField(dataValue.getKey(), (Integer) value);
+            else if (value instanceof Float) builder.addField(dataValue.getKey(), (Float) value);
+            else if (value instanceof Short) builder.addField(dataValue.getKey(), (Short) value);
+            else if (value instanceof Number) builder.addField(dataValue.getKey(), (Number) value);
+            else if (value instanceof String)
+                builder.addField(dataValue.getKey(), (String) value);
+            else return false;
+            //构建点
+            Point point = builder.build();
             log.info("插入点位数据:{}", point);
             //插入点
             result = influxDbUtil.insert(point);

iot-server/iot-server-data/src/main/java/com/shkpr/iot/server/data/service/impl/EquipmentServiceImpl.java

@@ -4,6 +4,7 @@ import com.shkpr.iot.common.core.domain.po.Destination;
 import com.shkpr.iot.common.core.domain.po.Equipment;
 import com.shkpr.iot.common.core.domain.po.PointRule;
 import com.shkpr.iot.common.core.domain.po.Template;
+import com.shkpr.iot.common.core.enums.ValueTypeEnum;
 import com.shkpr.iot.server.data.service.EquipmentService;
 import org.springframework.stereotype.Service;
 
@@ -50,21 +51,25 @@ public class EquipmentServiceImpl implements EquipmentService {
         pointRule1.setKey("ph");
         pointRule1.setFlagIndex("sensorId,267");
         pointRule1.setValueIndex("dataValue");
+        pointRule1.setValueType(ValueTypeEnum.SINGLE_FLOAT);
         //浊度
         PointRule pointRule2 = new PointRule();
         pointRule2.setKey("turbidity");
         pointRule2.setFlagIndex("sensorId,9");
         pointRule2.setValueIndex("dataValue");
+        pointRule2.setValueType(ValueTypeEnum.SINGLE_FLOAT);
         //余氯
         PointRule pointRule3 = new PointRule();
         pointRule3.setKey("chlorine");
         pointRule3.setFlagIndex("sensorId,264");
         pointRule3.setValueIndex("dataValue");
+        pointRule3.setValueType(ValueTypeEnum.SINGLE_FLOAT);
         //水温
         PointRule pointRule4 = new PointRule();
         pointRule4.setKey("temperature");
         pointRule4.setFlagIndex("sensorId,10");
         pointRule4.setValueIndex("dataValue");
+        pointRule4.setValueType(ValueTypeEnum.SINGLE_FLOAT);
         template.setRules(Arrays.asList(pointRule1, pointRule2, pointRule3, pointRule4));
         equipment.setTemplates(Collections.singletonList(template));