江津数据采集 更改采集逻辑和相关配置

dc3-gateway/src/main/java/io/github/pnoker/gateway/utils/KafkaConsumerDemo.java

@@ -141,7 +141,7 @@ public static void main(String[] args) {
     influxDB.setDatabase(database);
 
     // 查询数据
-    Query query = new Query("SELECT * FROM WaterMeter", database);
+    Query query = new Query("SELECT * FROM ThMeter", database);
     QueryResult result = influxDB.query(query);
 
     // 处理数据
@@ -172,28 +172,32 @@ public static void main(String[] args) {
 
                 String devId = (String) value.get(columns.indexOf("dev_id"));
                 String devId1 = (String) value.get(columns.indexOf("dev_id_1"));
-                Double flow_cur = (Double) value.get(columns.indexOf("flow_cur"));
-                Double flow_l_cur = (Double) value.get(columns.indexOf("flow_l_cur"));
-                Double flow_l_total_pos = (Double) value.get(columns.indexOf("flow_l_total_pos"));
-                Double flow_total_net = (Double) value.get(columns.indexOf("flow_total_net"));
-                Double flow_total_pos = (Double) value.get(columns.indexOf("flow_total_pos"));
-                Double flow_total_rev = (Double) value.get(columns.indexOf("flow_total_rev"));
-                Double press_cur = (Double) value.get(columns.indexOf("press_cur"));
+//                Double flow_cur = (Double) value.get(columns.indexOf("flow_cur"));
+//                Double flow_l_cur = (Double) value.get(columns.indexOf("flow_l_cur"));
+//                Double flow_l_total_pos = (Double) value.get(columns.indexOf("flow_l_total_pos"));
+//                Double flow_total_net = (Double) value.get(columns.indexOf("flow_total_net"));
+//                Double flow_total_pos = (Double) value.get(columns.indexOf("flow_total_pos"));
+//                Double flow_total_rev = (Double) value.get(columns.indexOf("flow_total_rev"));
+//                Double press_cur = (Double) value.get(columns.indexOf("press_cur"));
+                Double env_humidity = (Double) value.get(columns.indexOf("env_humidity"));
+                Double env_temperature = (Double) value.get(columns.indexOf("env_temperature"));
 
                 if (devId == null || devId.isEmpty()) {
                     devId = devId1; // 如果 dev_id 为空，则使用 dev_id_1 的值
                 }
 
-                Point point = Point.measurement("WaterMeter_temp")
+                Point point = Point.measurement("ThMeter_temp")
                         .time(timeNanos, TimeUnit.NANOSECONDS)
                         .tag("dev_id", devId)
-                        .addField("flow_cur", flow_cur)
-                        .addField("flow_l_cur", flow_l_cur)
-                        .addField("flow_l_total_pos", flow_l_total_pos)
-                        .addField("flow_total_net", flow_total_net)
-                        .addField("flow_total_pos", flow_total_pos)
-                        .addField("flow_total_rev", flow_total_rev)
-                        .addField("press_cur", press_cur)
+//                        .addField("flow_cur", flow_cur)
+////                        .addField("flow_l_cur", flow_l_cur)
+////                        .addField("flow_l_total_pos", flow_l_total_pos)
+////                        .addField("flow_total_net", flow_total_net)
+////                        .addField("flow_total_pos", flow_total_pos)
+////                        .addField("flow_total_rev", flow_total_rev)
+////                        .addField("press_cur", press_cur)
+                        .addField("env_humidity", env_humidity)
+                        .addField("env_temperature", env_temperature)
                         .build();
 
                 batchPoints.point(point);