mark for pull

2022-11-25 10:12:33 +08:00
parent 7abcafb93d
commit bd9276036b
15 changed files with 1002 additions and 45 deletions
--- a/ym-influx/src/main/java/com/cnbm/influx/config/InfluxClient.java
+++ b/ym-influx/src/main/java/com/cnbm/influx/config/InfluxClient.java
@@ -92,13 +92,25 @@ public enum InfluxClient {
    public void batchInsert(List<Event> events, String measurement){
        List<Point> list = new ArrayList<>();
        for(Event event:events){
-            Point point = Point.measurement(measurement)
-                    .addTag("transationId", event.getTransationId())
-                    .addTag("inspectionSheetId", event.getInspectionSheetId())
-                    .addTag("batchNum", event.getBatchNum().toString())
-                    .addTag("argName", event.getArgName())
-                    .addField("argValue", event.getArgValue())
-                    .time(event.getTime().toEpochMilli(), WritePrecision.MS);
+            Point point = null;
+            if(event.getBatchNum()==null){
+                point = Point.measurement(measurement)
+                        .addTag("transationId", event.getTransationId())
+                        .addTag("inspectionSheetId", event.getInspectionSheetId())
+                        .addTag("argName", event.getArgName())
+                        .addField("argValue", event.getArgValue())
+                        .time(event.getTime().toEpochMilli(), WritePrecision.MS);
+            }else {
+                point = Point.measurement(measurement)
+                        .addTag("transationId", event.getTransationId())
+                        .addTag("inspectionSheetId", event.getInspectionSheetId())
+                        .addTag("batchNum", event.getBatchNum().toString())
+                        .addTag("argName", event.getArgName())
+                        .addField("argValue", event.getArgValue())
+                        .time(event.getTime().toEpochMilli(), WritePrecision.MS);
+            }
+
+
            list.add(point);
        }
        writeApi.writePoints(list);
@@ -125,16 +137,31 @@ public enum InfluxClient {
        if(pageInfo!=null){
            flux += "|> limit(n: "+pageInfo.getSize()+", offset: "+(pageInfo.getCurrent()-1)* pageInfo.getSize()+")";
        }
+        return queryApi.query(flux);
+    }

+    public List<FluxTable> queryByGroup(QueryDataParam param){
+        String measurement = param.getMeasurement();
+        List<String> dropedTagNames = param.getDropedTagNames();
+        Range range = param.getRange();
+        String bucket = param.getBucket();
+        String tagName = param.getTag().getTagName();
+        String tagValue = param.getTag().getTagValue();
+        String groupName = param.getGroupName();
+        PageInfo pageInfo = param.getPageInfo();

-//        List<FluxTable> tables = queryApi.query(flux);
-//        for (FluxTable fluxTable : tables) {
-//            List<FluxRecord> records = fluxTable.getRecords();
-//            for (FluxRecord fluxRecord : records) {
-//                System.out.println("time: "+fluxRecord.getTime() +"   key:"+fluxRecord.getField()+"   value: " + fluxRecord.getValueByKey("_value")+"   measurement: " + fluxRecord.getMeasurement());
-//
-//            }
-//        }
+        String flux = "from(bucket:\""+bucket+"\")";
+        flux += "|> range(start: "+range.getBegin()+",stop:"+range.getEnd()+")";
+        flux += "|> filter(fn: (r) => r[\"_measurement\"] == \""+measurement+"\")";
+        flux += "|> filter(fn: (r) => r[\""+tagName+"\"] == \""+tagValue+"\")";
+        flux += "|> group(columns: [\""+groupName+"\"], mode: \"except\")";
+        for(String dropName:dropedTagNames){
+            flux += "|> drop(columns: [\""+dropName+"\"])";
+        }
+        flux += "|> sort(columns: [\"_time\"], desc: true)";
+        if(pageInfo!=null){
+            flux += "|> limit(n: "+pageInfo.getSize()+", offset: "+(pageInfo.getCurrent()-1)* pageInfo.getSize()+")";
+        }
        return queryApi.query(flux);
    }

--- a/ym-influx/src/main/java/com/cnbm/influx/constant/Constant.java
+++ b/ym-influx/src/main/java/com/cnbm/influx/constant/Constant.java
@@ -18,7 +18,7 @@ public class Constant {
    public static final LogLevel readTimeout = LogLevel.BODY;
    public static final LogLevel writeTimeout = LogLevel.BODY;
    public static final LogLevel connectTimeout = LogLevel.BODY;
-    public static final String measurement = "WeightHeiHei";
+    public static final String measurement = "Weights";


 }
--- a/ym-influx/src/main/java/com/cnbm/influx/controller/S7DemoController.java
+++ b/ym-influx/src/main/java/com/cnbm/influx/controller/S7DemoController.java
@@ -1,6 +1,8 @@
 package com.cnbm.influx.controller;

 import com.cnbm.common.spc.util.DataUtils;
+import com.cnbm.common.utils.Result;
+import com.cnbm.common.vo.R;
 import com.cnbm.influx.config.InfluxClient;
 import com.cnbm.influx.param.PageInfo;
 import com.cnbm.influx.param.QueryDataParam;
@@ -43,6 +45,27 @@ public class S7DemoController {
            event.setArgName("LTWeight");
            Double d = r.nextDouble() * 2.5  + 66;
            event.setArgValue(d);
+
+            list.add(event);
+        }
+        InfluxClient.Client.batchInsert(list,"Weight");
+    }
+
+    @PostMapping("/forTestInsertBatch")
+    public void forTestInsertBatch() throws InterruptedException {
+        List<Event> list = new ArrayList<>();
+        Random r = new Random();
+
+        for(int i=0;i<99;i++){
+            Thread.sleep(10);
+            Event event = new Event();
+            event.setTime(Instant.now());
+            event.setTransationId("forbatch"+i);
+            event.setArgName("LTWeight");
+            Double d = r.nextDouble() * 2.5  + 66;
+            event.setArgValue(d);
+            event.setBatchNum(4);
+            event.setInspectionSheetId(i+"");
            list.add(event);
        }
        InfluxClient.Client.batchInsert(list,"Weight");
@@ -71,6 +94,8 @@ public class S7DemoController {
        InfluxClient.Client.batchInsert(list,"WeightHei");
    }

+
+
    @PostMapping("/insertBatchJYDForTest")
    public void insertBatchJYDForTest() throws InterruptedException {
        List<Event> list = new ArrayList<>();
@@ -88,13 +113,39 @@ public class S7DemoController {
            }else {
                event.setArgValue(new Double(1));
            }
-
            event.setInspectionSheetId(i+"");

            event.setBatchNum(i);
            list.add(event);
        }
-        InfluxClient.Client.batchInsert(list,"WeightHeiHei");
+        InfluxClient.Client.batchInsert(list,"Weights");
+    }
+
+    @PostMapping("/insertBatchForNew")
+    public void insertBatchForNew() throws InterruptedException {
+        List<Event> list = new ArrayList<>();
+        Random r = new Random();
+
+        for(int i=0;i<999;i++){
+            Thread.sleep(10);
+            Event event = new Event();
+            event.setTime(new Date().toInstant());
+            event.setTransationId("asas"+i);
+            event.setArgName("LiuWeight");
+//            int i1 = r.nextInt(10);
+//            if(i1<4){
+//                event.setArgValue(new Double(0));
+//            }else {
+//                event.setArgValue(new Double(1));
+//            }
+            Double d = r.nextDouble() * 2.5  + 66;
+            event.setArgValue(d);
+            event.setInspectionSheetId(i+"");
+
+            event.setBatchNum(i);
+            list.add(event);
+        }
+        InfluxClient.Client.batchInsert(list,"Weights");
    }

    /**
@@ -131,7 +182,6 @@ public class S7DemoController {
            List<FluxRecord> records = fluxTable.getRecords();
            for (FluxRecord fluxRecord : records) {
                System.out.println("value: " + fluxRecord.getValueByKey("_value"));
-
            }
        }
        System.out.println();
@@ -157,7 +207,6 @@ public class S7DemoController {
            List<FluxRecord> records = fluxTable.getRecords();
            for (FluxRecord fluxRecord : records) {
                System.out.println("value: " + fluxRecord.getValueByKey("_value"));
-
            }
        }
    }
--- a/ym-influx/src/main/java/com/cnbm/influx/param/QueryDataParam.java
+++ b/ym-influx/src/main/java/com/cnbm/influx/param/QueryDataParam.java
@@ -29,5 +29,6 @@ public class QueryDataParam extends BaseParam{
    //查询的时候，需要忽略的字段。(transationId是唯一标识会对 最终的查询结果集产生影响)
    private List<String> dropedTagNames;
    private String bucket;
+    private String groupName;

 }