influxdb 进行简单封装

amos-boot-system-jxiop/amos-boot-module-jxiop-api/src/main/java/com/yeejoin/amos/boot/module/jxiop/api/dto/IndexDto.java

@@ -12,8 +12,8 @@ import java.util.Date;
 //风机指标信息
 @Data
 public class IndexDto {
-    private String gateway;// '网关',
-    private String indicatorValue;// '指标值', 取valueLable
-    private String indexAddress;//  指标地址
-    private Date time;// '上报时间',
+    private String gatewayId;// '网关',
+    private String valueLabel;// '指标值', 取valueLable
+    private String address;//  指标地址
+    private Date createdTime;// '上报时间',
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-monitor-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/controller/DemoController.java

@@ -3,22 +3,30 @@ package com.yeejoin.amos.boot.module.jxiop.biz.controller;
 
 import com.yeejoin.amos.boot.biz.common.controller.BaseController;
 import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
+import com.yeejoin.amos.boot.module.jxiop.api.dto.IndexDto;
 import com.yeejoin.amos.boot.module.jxiop.api.entity.StationBasic;
 import com.yeejoin.amos.boot.module.jxiop.api.mapper.StationBasicMapper;
 import com.yeejoin.amos.boot.module.jxiop.biz.entity.Test;
 import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.SjglZsjZsbtzServiceImpl;
 import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.StationBasicServiceImpl;
 import com.yeejoin.amos.boot.module.jxiop.biz.service.impl.TestServiceImpl;
+import com.yeejoin.amos.boot.module.jxiop.biz.utils.InfluxDButils;
 import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
 import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
+import io.swagger.models.auth.In;
+import net.logstash.logback.encoder.org.apache.commons.lang3.StringUtils;
 import org.influxdb.dto.QueryResult;
+import org.springframework.beans.BeanWrapperImpl;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 import org.typroject.tyboot.core.foundation.enumeration.UserType;
 import org.typroject.tyboot.core.restful.doc.TycloudOperation;
 
+import java.util.ArrayList;
+import java.util.Date;
 import java.util.List;
+import java.util.Map;
 
 @RestController
 @Api(tags = "Demo-test")
@@ -39,6 +47,8 @@ public class DemoController extends BaseController {
     InfluxDbConnection influxDbConnection;
     @Autowired
     SjglZsjZsbtzServiceImpl sjglZsjZsbtzServiceImpl;
+    @Autowired
+    InfluxDButils influxDButils;
 
 
     @TycloudOperation(needAuth = false, ApiLevel = UserType.AGENCY)
@@ -57,19 +67,9 @@ public class DemoController extends BaseController {
     @ApiOperation(value = "demo测试influxdb")
     @GetMapping("/test3")
     public void demoTest2() {
-        QueryResult query =influxDbConnection.query("select * from iot_data  where time> now()-1s");
-        List<QueryResult.Result >  queryResults=query.getResults();
-        for (int i = 0; i <queryResults.size() ; i++) {
-           QueryResult.Result  result = queryResults.get(i);
-          List list =result.getSeries();
-            for (int j = 0; j <list.size() ; j++) {
-                QueryResult.Series series  = (QueryResult.Series) list.get(j);
-                System.out.println(series.getName());
-                System.out.println(series.getColumns());
-                System.out.println(series.getValues());
-                System.out.println(series.getTags());
-            }
-        }
+        String sql = "SELECT createdTime ,gatewayId,address,valueLabel FROM \"iot_data\" WHERE equipmentSpecificName=~/.*27风机.*/  and  gatewayId='1668801435891929089'    GROUP BY equipmentIndex    ORDER BY time  desc  LIMIT 600";
+        List<IndexDto> list=influxDButils.getListData(sql,IndexDto.class);
+        System.out.println(list.size());
     }
     @TycloudOperation(needAuth = false, ApiLevel = UserType.AGENCY)
     @ApiOperation(value = "测试数据同步逻辑")

amos-boot-system-jxiop/amos-boot-module-jxiop-monitor-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/service/impl/StationDataTaskImpl.java

@@ -8,6 +8,7 @@ import com.yeejoin.amos.boot.module.jxiop.api.entity.StationBasic;
 import com.yeejoin.amos.boot.module.jxiop.api.service.IStationDataTask;
 import com.yeejoin.amos.boot.module.jxiop.biz.entity.SjglZsjZsbtz;
 import com.yeejoin.amos.boot.module.jxiop.biz.entity.TpriDmpDatabook;
+import com.yeejoin.amos.boot.module.jxiop.biz.utils.InfluxDButils;
 import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
 import org.influxdb.dto.QueryResult;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -34,6 +35,8 @@ public class StationDataTaskImpl {
     InfluxDbConnection influxDbConnection;
     @Autowired
     MonitorFanIndicatorImpl monitorFanIndicatorImpl;
+    @Autowired
+    InfluxDButils influxDButils;
 
 
     @Value("${station.section}")
@@ -73,26 +76,24 @@ public class StationDataTaskImpl {
 
     @Async("jxiopAsyncExecutor")
     public List<IndexDto> getIndexDto(DeviceDto deviceDto) {
-
+        String querysql ="SELECT createdTime ,gatewayId,address,valueLabel FROM iot_data WHERE equipmentSpecificName=~/.*"+deviceDto.getNumberName()+".*/  and  gatewayId='"+deviceDto.getGateway()+"'    GROUP BY equipmentIndex    ORDER BY time  desc  LIMIT 600";
         //每个分机的指标数据
-        List<IndexDto> list = null;
-        QueryResult query = influxDbConnection.query("SELECT * FROM iot_data where \"equipmentSpecificName\" =~/.*"+ deviceDto.getNumberName()+".*/  and \"gatewayId\" =~/" + deviceDto.getGateway() + "/   group by \"equipmentsIdx\" limit 1");
-        List<QueryResult.Result> queryResults = query.getResults();
-        for (int i = 0; i < queryResults.size(); i++) {
-            QueryResult.Result result = queryResults.get(i);
-            List seriesList = result.getSeries();
-            for (int j = 0; j < list.size(); j++) {
-                QueryResult.Series series = (QueryResult.Series) seriesList.get(j);
-                System.out.println(series.getName());
-                System.out.println(series.getColumns());
-                System.out.println(series.getValues());
-                System.out.println(series.getTags());
+        List<IndexDto> list = influxDButils.getListData(querysql,IndexDto.class);
+        if(list!=null&&list.size()>0){
+            //对数据切片处理 默认创建
+            if(stationSection>list.size()||!isok){
+                monitorFanIndicatorImpl.UpdateMonitorFanIndicator(list);
+            }else{
+                //进行分片处理
+                List<List<IndexDto>>  listfp= Lists.partition(list,list.size()%stationSection);
+                for (List<IndexDto> indexDtos : listfp) {
+                    //每个分级信息处理
+                    monitorFanIndicatorImpl.UpdateMonitorFanIndicator(indexDtos);
+                }
             }
         }
-
         return list;
     }
-
     @Async("jxiopAsyncExecutor")
     public void UpdateMonitorFanIndicator(List<IndexDto> list){
         if(list!=null&&list.size()>0){
@@ -110,4 +111,5 @@ public class StationDataTaskImpl {
         }
     }
 
+
 }

amos-boot-system-jxiop/amos-boot-module-jxiop-monitor-biz/src/main/java/com/yeejoin/amos/boot/module/jxiop/biz/utils/InfluxDButils.java

+package com.yeejoin.amos.boot.module.jxiop.biz.utils;
+
+import com.yeejoin.amos.component.influxdb.InfluxDbConnection;
+import org.influxdb.dto.QueryResult;
+import org.springframework.beans.BeanWrapperImpl;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+@Component
+public class InfluxDButils {
+    @Autowired
+    InfluxDbConnection influxDbConnection;
+
+    public <T> List<T> getListData(String sql, Class<T> clazz) {
+        List<T> list = new ArrayList<>();
+        try {
+            QueryResult query = influxDbConnection.query(sql);
+            List<QueryResult.Result> queryResults = query.getResults();
+            for (QueryResult.Result result : queryResults) {
+                List<QueryResult.Series> series = result.getSeries();
+                if (series == null) {
+                    continue;
+                }
+                for (QueryResult.Series serie : series) {
+                    List<List<Object>> values = serie.getValues();
+                    List<String> columns = serie.getColumns();
+                    for (int i = 0; i < values.size(); ++i) {
+                        T object = clazz.newInstance();
+                        BeanWrapperImpl bean = new BeanWrapperImpl(object);
+                        for (int j = 0; j < columns.size(); ++j) {
+                            String k = columns.get(j);
+                            Object v = values.get(i).get(j);
+                            if ("time".equals(k)) {
+                                continue;
+                            } else if ("createdTime".equals(k)) {
+                                Date date = DateUtil.getLongDate(v.toString());
+                                bean.setPropertyValue(k, date);
+                            } else {
+                                bean.setPropertyValue(k, v);
+                            }
+                        }
+                        list.add(object);
+                    }
+                }
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        return list;
+    }
+}