fit():优化项目依赖

pull/21/head
yanchouchou 2024-10-07 20:10:56 +08:00
parent 33f0a863c8
commit d4fc81a7a1
207 changed files with 1490 additions and 29 deletions

View File

@ -0,0 +1,32 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-common</artifactId>
<version>3.6.3</version>
</parent>
<description>
cloud-common-cache 缓存基准
</description>
<artifactId>cloud-common-cache</artifactId>
<properties>
<maven.compiler.source>23</maven.compiler.source>
<maven.compiler.target>23</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<!-- redis 缓存模块-->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-redis</artifactId>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,63 @@
package com.muyu.common.cache;
import com.muyu.common.core.web.domain.BaseEntity;
import com.muyu.common.redis.service.RedisService;
import org.springframework.beans.factory.annotation.Autowired;
import java.util.concurrent.TimeUnit;
/**
*
* * @className: CacheAbsBasic
* * @author: Yang 🦅
* * @date: 2024/9/29 16:08
* * @Version: 1.0
* * @description:
*/
public abstract class CacheAbsBasic <K, V> implements CacheBasic<K, V>{
@Autowired
private RedisService redisService;
@Override
public void put(K key, V value) {
try {
redisService.setCacheObject(encode(key), value,30L,TimeUnit.MINUTES);
} catch (Exception e) {
throw new RuntimeException("运行时异常,异常信息为:{}"+e.getMessage());
}
}
@Override
public V get(K key) {
try {
return redisService.getCacheObject(encode(key));
} catch (Exception e) {
throw new RuntimeException("运行时异常,异常信息为:{}"+e.getMessage());
}
}
@Override
public void remove(K key) {
try {
redisService.deleteObject(encode(key));
} catch (Exception e) {
throw new RuntimeException("运行时异常,异常信息为:{}"+e.getMessage());
}
}
@Override
public boolean hashKey(K key){
Boolean b = false;
try {
b = redisService.hasKey(encode(key));
} catch (Exception e) {
throw new RuntimeException("运行时异常,异常信息为:{}"+e.getMessage());
}
return b;
}
}

View File

@ -0,0 +1,24 @@
package com.muyu.common.cache;
import org.springframework.data.redis.core.TimeoutUtils;
import java.util.Date;
import java.util.concurrent.TimeUnit;
/**
*
* * @className: CacheBasic
* * @author: Yang 🦅
* * @date: 2024/9/29 16:08
* * @Version: 1.0
* * @description:
*/
public interface CacheBasic<K, V> extends PrimaryKeyBasic<K> {
void put(K key, V value);
V get(K key);
void remove(K key);
boolean hashKey(K key);
}

View File

@ -0,0 +1,36 @@
package com.muyu.common.cache;
/**
*
* * @className: PrimaryKeyBasic
* * @author: Yang 🦅
* * @date: 2024/9/29 16:08
* * @Version: 1.0
* * @description:
*/
public interface PrimaryKeyBasic <K>{
/**
*
* @return
*/
public String keyPre();
/**
*
* @param key
* @return
*/
public default String encode(K key){
return keyPre() + key.toString();
}
/**
*
* @param key
* @return
*/
public default K decode(String key) {
return (K) key.substring(keyPre().length());
}
}

View File

@ -132,10 +132,10 @@
</dependency>
<!-- Java Specification Requests 标准库-->
<!-- <dependency>-->
<!-- <groupId>javax.annotation</groupId>-->
<!-- <artifactId>jsr250-api</artifactId>-->
<!-- </dependency>-->
<!-- <dependency>-->
<!-- <groupId>javax.annotation</groupId>-->
<!-- <artifactId>jsr250-api</artifactId>-->
<!-- </dependency>-->
<dependency>
<groupId>javax.annotation</groupId>
@ -165,5 +165,4 @@
<artifactId>hutool-all</artifactId>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,40 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-common</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>cloud-common-iotdb</artifactId>
<description>
cloud-common-iotdb 时序性数据存储服务
</description>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
<version>1.3.1</version>
</dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,149 @@
package com.muyu.common.iotdb.config;
import com.alibaba.fastjson2.JSONObject;
import lombok.extern.log4j.Log4j2;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.rpc.IoTDBConnectionException;
import org.apache.iotdb.rpc.StatementExecutionException;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
/**
* @Author WangXin
* @Data 2024/9/30
* @Description IotDBSessionConfig
* @Version 1.0.0
*/
@Log4j2
@Component
@Configuration
public class IotDBSessionConfig {
@Value("${spring.iotdb.username:root}")
private String username;
@Value("${spring.iotdb.password:root}")
private String password;
@Value("${spring.iotdb.ip:127.0.0.1}")
private String ip;
@Value("${spring.iotdb.port:6667}")
private int port;
@Value("${spring.iotdb.maxSize:10}")
private int maxSize;
private static SessionPool sessionPool;
/**
* IotDBSession
* @return iotDBSession
*/
public SessionPool getSessionPool() {
if (sessionPool == null) {
sessionPool = new SessionPool(ip, port, username, password, maxSize);
}
return sessionPool;
}
/**
*
* @param deviceId
* @param time
* @param measurements
* @param values
*/
public void insertRecord(SessionPool sessionPool,String deviceId, long time, List<String> measurements, List<String> values) {
try {
log.info("iotdb数据入库device_id:[{}], measurements:[{}], values:[{}]", deviceId, measurements, values);
sessionPool.insertRecord(deviceId, time, measurements, values);
} catch (Exception e) {
log.error("IotDBSession insertRecord失败: deviceId={}, time={}, measurements={}, values={}, error={}",
deviceId, time, measurements, values, e.getMessage());
}
}
public SessionDataSet selectRecord(SessionPool sessionPool,String sql) {
log.info("iotdb数据查询sql:[{}]",sql);
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb SQL查询sql:[{}]", sql);
sessionDataSetWrapper = sessionPool.executeQueryStatement(sql);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
throw new RuntimeException(e);
}
}
public static void main(String[] args) {
SessionPool sessionPool = new SessionPool("127.0.0.1", 6667, "root", "root", 10);
String ROOT_DATA_DATAJSON = "root.car.data.datajson";
String SELECT_ROOT_DATA_DATAJSON_DATASOURCE = "select * from root.car.data.datajson";
String jsonValue = """
{
"name": "张三",
"age": 28,
"email": "zhangsan@example.com",
"isStudent": false,
"hobbies": ["阅读", "旅行", "编程"],
"address": {
"street": "长安街100号",
"city": "北京",
"postalCode": "100000"
}
}
""";
IotDBSessionConfig iotDBSessionConfig = new IotDBSessionConfig();
List<String> values = new ArrayList<>();
values.add(jsonValue);
ArrayList<String> objects = new ArrayList<>();
objects.add("datasource");
iotDBSessionConfig.insertRecord(sessionPool,ROOT_DATA_DATAJSON,System.currentTimeMillis(),objects,values);
SessionDataSet sessionDataSet = iotDBSessionConfig.selectRecord(sessionPool,SELECT_ROOT_DATA_DATAJSON_DATASOURCE);
HashMap<Long, Map<String, String>> longMapHashMap = new HashMap<>();
try {
while (sessionDataSet.hasNext()){
RowRecord next = sessionDataSet.next();
long timestamp = next.getTimestamp();
Map<String, String> fieldMap = new HashMap<>();
for (Field field : next.getFields()) {
TSDataType dataType = field.getDataType();
String stringValue = field.getStringValue();
fieldMap.put(dataType.name(), stringValue);
}
longMapHashMap.put(timestamp, fieldMap);
}
} catch (StatementExecutionException e) {
throw new RuntimeException(e);
} catch (IoTDBConnectionException e) {
throw new RuntimeException(e);
}
log.info("数据为:{}", JSONObject.toJSONString(longMapHashMap));
}
}

View File

@ -0,0 +1,29 @@
package com.muyu.common.iotdb.domain;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import java.util.List;
/**
* @Author WangXin
* @Data 2024/9/29
* @Description
* @Version 1.0.0
*/
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class EventActuate {
/**
* json
*/
private String jsonData;
/**
* key
*/
private List<String> eventKeys;
}

View File

@ -0,0 +1,40 @@
package com.muyu.common.iotdb.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.ArrayList;
import java.util.List;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class InsertDataDTO {
private Float temperature;
private String hardware;
private Boolean status;
public InsertDataDTO buildOne() {
InsertDataDTO insertDataDTO = new InsertDataDTO();
insertDataDTO.setHardware("ss");
insertDataDTO.setStatus(true);
insertDataDTO.setTemperature(12.0F);
return insertDataDTO;
}
public List<InsertDataDTO> buildList() {
List<InsertDataDTO> insertDataDTOS = new ArrayList<>();
int buildNum = 10;
for (int i = 0; i < buildNum; i++) {
InsertDataDTO insertDataDTO = new InsertDataDTO();
insertDataDTO.setHardware(i % 2 == 0 ? "pp" + i : null);
insertDataDTO.setStatus(i % 2 == 0);
insertDataDTO.setTemperature(12.0F + i);
insertDataDTOS.add(insertDataDTO);
}
return insertDataDTOS;
}
}

View File

@ -0,0 +1,20 @@
package com.muyu.common.iotdb.domain;
import com.muyu.common.iotdb.domain.dto.IotDbRecordAble;
import lombok.Data;
import lombok.EqualsAndHashCode;
@EqualsAndHashCode(callSuper = true)
@Data
public class ResultEntity extends IotDbRecordAble {
private Float temperature;
private String hardware;
private Boolean status;
private String time;
}

View File

@ -0,0 +1,18 @@
package com.muyu.common.iotdb.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class TestDataType {
private Float temperature;
private String hardware;
private Boolean status;
private Double testDouble;
private Long testLong;
}

View File

@ -0,0 +1,14 @@
package com.muyu.common.iotdb.domain.dto;
import lombok.Data;
/**
* @Author WangXin
* @Data 2024/9/30
* @Description IotDBServiceImpl
* @Version 1.0.0
*/
@Data
public class IotDbRecordAble {
}

View File

@ -0,0 +1,23 @@
package com.muyu.common.iotdb.domain.dto;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.util.List;
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class MeasurementSchemaValuesDTO {
private List<MeasurementSchema> schemaList;
private List<Object> values;
private List<Integer> valueIsNullIndex;
}

View File

@ -0,0 +1,105 @@
package com.muyu.common.iotdb.service;
import com.muyu.common.iotdb.domain.dto.IotDbRecordAble;
import com.muyu.common.iotdb.domain.dto.MeasurementSchemaValuesDTO;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import java.util.List;
import java.util.Map;
/**
* @Author WangXin
* @Data 2024/9/28
* @Description IotDBServiceImpl
* @Version 1.0.0
*/
public interface IotDBService {
void insertTablet(Tablet tablet);
void insertTablets(Map<String, Tablet> tablets);
void insertStringRecord(String deviceId, long time, List<String> measurements, List<String> values);
void insertRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values);
void insertStringRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList);
void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList);
void insertStringRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList);
void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList);
void deleteData(String path, long endTime);
void deleteData(List<String> paths, long endTime);
SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime, long timeOut);
<T> List<T> executeRawDataQuery(List<String> paths, long startTime, long endTime, long timeOut, Class<? extends IotDbRecordAble> clazz);
SessionDataSet executeLastDataQuery(List<String> paths, long lastTime);
<T> List<T> executeLastDataQuery(List<String> paths, long lastTime, Class<? extends IotDbRecordAble> clazz);
SessionDataSet executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes);
<T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes, Class<? extends IotDbRecordAble> clazz);
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations);
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime);
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval);
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval, long slidingStep);
SessionDataSet executeQueryStatement(String sql);
/**
* SQL
*
* @param sql
*/
void executeNonQueryStatement(String sql);
/**
*
*
* @param sessionDataSet
* @param titleList
*/
List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> columnNames);
/**
*
*
* @param sessionDataSet
* @param titleList
* @param clazz
* @param <T>
* @return
*/
<T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList, Class<? extends IotDbRecordAble> clazz);
/**
* MeasurementSchemas
*
* @param object
* @return
*/
List<MeasurementSchema> buildMeasurementSchemas(Object object);
/**
* MeasurementSchemaValuesDTO
*
* @param object
* @return
*/
MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object object);
}

View File

@ -0,0 +1,711 @@
package com.muyu.common.iotdb.service.impl;
import com.alibaba.fastjson2.JSON;
import com.muyu.common.iotdb.config.IotDBSessionConfig;
import com.muyu.common.iotdb.domain.dto.IotDbRecordAble;
import com.muyu.common.iotdb.domain.dto.MeasurementSchemaValuesDTO;
import com.muyu.common.iotdb.service.IotDBService;
import jakarta.annotation.Resource;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
import org.apache.iotdb.isession.SessionDataSet;
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
import org.apache.iotdb.session.pool.SessionPool;
import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
import org.apache.iotdb.tsfile.read.common.Field;
import org.apache.iotdb.tsfile.read.common.RowRecord;
import org.apache.iotdb.tsfile.write.record.Tablet;
import org.apache.iotdb.tsfile.write.schema.MeasurementSchema;
import org.springframework.stereotype.Service;
import java.lang.reflect.Type;
import java.util.*;
import java.util.stream.Collectors;
/**
* @Author WangXin
* @Data 2024/9/28
* @Description IotDBServiceImpl
* @Version 1.0.0
*/
@Service
@Slf4j
public class IotDBServiceImpl implements IotDBService {
@Resource
private IotDBSessionConfig iotDBSessionConfig;
/**
*
*
* @param tablet
*/
@Override
public void insertTablet(Tablet tablet) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库tablet:[{}]", tablet);
sessionPool.insertTablet(tablet);
} catch (Exception e) {
log.error("IotDBSession insertTablet失败: tablet={}, error={}", tablet, e.getMessage());
}
}
/**
*
*
* @param tablets
*/
@Override
public void insertTablets(Map<String, Tablet> tablets) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库tablets:[{}]", tablets);
sessionPool.insertTablets(tablets);
} catch (Exception e) {
log.error("IotDBSession insertTablets失败: tablets={}, error={}", tablets, e.getMessage());
}
}
/**
* string
*
* @param deviceId root.ln.wf01.wt01
* @param time
* @param measurements
* @param values
*/
@Override
public void insertStringRecord(String deviceId, long time, List<String> measurements, List<String> values) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库device_id:[{}], measurements:[{}], values:[{}]", deviceId, measurements, values);
sessionPool.insertRecord(deviceId, time, measurements, values);
} catch (Exception e) {
log.error("IotDBSession insertRecord失败: deviceId={}, time={}, measurements={}, values={}, error={}",
deviceId, time, measurements, values, e.getMessage());
}
}
/**
*
*
* @param deviceId root.ln.wf01.wt01
* @param time
* @param measurements
* @param types
* @param values
*/
@Override
public void insertRecord(String deviceId, long time, List<String> measurements, List<TSDataType> types, List<Object> values) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库device_id:[{}], measurements:[{}], types:[{}], values:[{}]", deviceId, measurements, types, values);
sessionPool.insertRecord(deviceId, time, measurements, types, values);
} catch (Exception e) {
log.error("IotDBSession insertRecordHasTypes失败: deviceId={}, time={}, measurements={},types={}, values={}, error={}",
deviceId, time, measurements, types, values, e.getMessage());
}
}
/**
* string
*
* @param deviceIds root.ln.wf01.wt01
* @param times
* @param measurementsList
* @param valuesList
*/
@Override
public void insertStringRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库deviceIds:[{}], measurementsList:[{}], valuesList:[{}]", deviceIds, measurementsList, valuesList);
sessionPool.insertRecords(deviceIds, times, measurementsList, valuesList);
} catch (Exception e) {
log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, valuesList={}, error={}",
deviceIds, times, measurementsList, valuesList, e.getMessage());
}
}
/**
*
*
* @param deviceIds root.ln.wf01.wt01
* @param times
* @param measurementsList
* @param typesList
* @param valuesList
*/
@Override
public void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库deviceIds:[{}], measurementsList:[{}], typesList[{}], valuesList:[{}]", deviceIds, measurementsList, typesList, valuesList);
sessionPool.insertRecords(deviceIds, times, measurementsList, typesList, valuesList);
} catch (Exception e) {
log.error("IotDBSession insertRecords失败: deviceIds={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}",
deviceIds, times, measurementsList, typesList, valuesList, e.getMessage());
}
}
/**
* string
*
* @param deviceId root.ln.wf01.wt01
* @param times
* @param measurementsList
* @param valuesList
*/
@Override
public void insertStringRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<String>> valuesList) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库deviceId:[{}], measurementsList:[{}], valuesList:[{}]", deviceId, measurementsList, valuesList);
sessionPool.insertStringRecordsOfOneDevice(deviceId, times, measurementsList, valuesList);
} catch (Exception e) {
log.error("IotDBSession insertStringRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, valuesList={}, error={}",
deviceId, times, measurementsList, valuesList, e.getMessage());
}
}
/**
*
*
* @param deviceId root.ln.wf01.wt01
* @param times
* @param measurementsList
* @param typesList
* @param valuesList
*/
@Override
public void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList, List<List<TSDataType>> typesList, List<List<Object>> valuesList) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据入库deviceId:[{}], measurementsList:[{}], typesList[{}], valuesList:[{}]", deviceId, measurementsList, typesList, valuesList);
sessionPool.insertRecordsOfOneDevice(deviceId, times, measurementsList, typesList, valuesList);
} catch (Exception e) {
log.error("IotDBSession insertRecordsOfOneDevice失败: deviceId={}, times={}, measurementsList={}, typesList=[],valuesList={}, error={}", deviceId, times, measurementsList, typesList, valuesList, e.getMessage());
}
}
/**
*
*
* @param path root.ln.wf01.wt01.temperature
* @param endTime
*/
@Override
public void deleteData(String path, long endTime) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据删除path:[{}], endTime:[{}]", path, endTime);
sessionPool.deleteData(path, endTime);
} catch (Exception e) {
log.error("IotDBSession deleteData失败: deviceId={}, times={},error={}", path, endTime, e.getMessage());
}
}
/**
*
*
* @param paths root.ln.wf01.wt01.temperature
* @param endTime
*/
@Override
public void deleteData(List<String> paths, long endTime) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb数据删除paths:[{}], endTime:[{}]", paths, endTime);
sessionPool.deleteData(paths, endTime);
} catch (Exception e) {
log.error("IotDBSession deleteData失败: paths={}, times={},error={}", paths, endTime, e.getMessage());
}
}
/**
*
*
* @param paths root.ln.wf01.wt01.temperature
* @param startTime
* @param endTime
* @param outTime
* @return SessionDataSet (Time,paths)
*/
@Override
public SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb数据查询paths:[{}], startTime[{}], endTime:[{}],outTime:[{}]", paths, startTime, endTime, outTime);
sessionDataSetWrapper = sessionPool.executeRawDataQuery(paths, startTime, endTime, outTime);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeRawDataQuery失败: paths={}, startTime[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
*
*
* @param paths root.ln.wf01.wt01.temperature
* @param startTime
* @param endTime
* @param outTime
* @param clazz
* @param <T>
* @return
*/
@Override
public <T> List<T> executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime, Class<? extends IotDbRecordAble> clazz) {
SessionDataSet sessionDataSet = executeRawDataQuery(paths, startTime, endTime, outTime);
List<String> columnNames = sessionDataSet.getColumnNames();
List<T> resultEntities = null;
try {
resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
} catch (Exception e) {
log.error("IotDBSession executeRawDataQuery失败: paths={}, startTime[{}], endTime:[{}],outTime:[{}],error={}", paths, startTime, endTime, outTime, e.getMessage());
}
return resultEntities;
}
/**
* ()
*
* @param paths root.ln.wf01.wt01.temperature
* @param lastTime
* @return SessionDataSet
*/
@Override
public SessionDataSet executeLastDataQuery(List<String> paths, long lastTime) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb数据查询paths:[{}], lastTime:[{}]", paths, lastTime);
sessionDataSetWrapper = sessionPool.executeLastDataQuery(paths, lastTime);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeLastDataQuery失败: paths={}, lastTime[{}], error={}", paths, lastTime, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* ()
*
* @param <T>
* @param paths root.ln.wf01.wt01.temperature
* @param lastTime
* @param clazz
* @return
*/
@Override
public <T> List<T> executeLastDataQuery(List<String> paths, long lastTime, Class<? extends IotDbRecordAble> clazz) {
SessionDataSet sessionDataSet = executeLastDataQuery(paths, lastTime);
List<String> columnNames = sessionDataSet.getColumnNames();
List<T> resultEntities = null;
try {
resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
} catch (Exception e) {
log.error("IotDBSession executeLastDataQuery失败: paths={}, lastTime[{}], error={}", paths, lastTime, e.getMessage());
}
return resultEntities;
}
/**
*
*
* @param db root.ln.wf01
* @param device root.ln.wf01.wt01
* @param sensors temperaturestatus()
* @param isLegalPathNodes true()
* @return SessionDataSet
*/
@Override
public SessionDataSet executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb数据查询db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}]", db, device, sensors, isLegalPathNodes);
sessionDataSetWrapper = sessionPool.executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* @param db root.ln.wf01
* @param device root.ln.wf01.wt01
* @param sensors temperaturestatus()
* @param isLegalPathNodes true()
* @param clazz
* @param <T>
* @return
*/
@Override
public <T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors, boolean isLegalPathNodes, Class<? extends IotDbRecordAble> clazz) {
SessionDataSet sessionDataSet = executeLastDataQueryForOneDevice(db, device, sensors, isLegalPathNodes);
List<String> columnNames = sessionDataSet.getColumnNames();
List<T> resultEntities = null;
try {
resultEntities = packagingObjectData(sessionDataSet, columnNames, clazz);
} catch (Exception e) {
log.error("IotDBSession executeLastDataQueryForOneDevice失败: db:[{}], device:[{}],sensors:[{}], isLegalPathNodes:[{}], error={}", db, device, sensors, isLegalPathNodes, e.getMessage());
}
return resultEntities;
}
/**
*
*
* @param paths root.ln.wf01.wt01.temperature
* @param aggregations TAggregationType.SUM,TAggregationType.COUNT
* @return SessionDataSet
*/
@Override
public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb聚合查询paths:[{}], aggregations[{}]", paths, aggregations);
sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations[{}] ,error={}", paths, aggregations, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* ()
*
* @param paths root.ln.wf01.wt01.temperature
* @param aggregations TAggregationType.SUM,TAggregationType.COUNT
* @param startTime ()
* @param endTime
* @return SessionDataSet
*/
@Override
public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb聚合查询paths:[{}], aggregations[{}],startTime[{}], endTime:[{}]", paths, aggregations, startTime, endTime);
sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations[{}] ,startTime[{}], endTime:[{}],error={}", paths, aggregations, startTime, endTime, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* ()
*
* @param paths root.ln.wf01.wt01.temperature
* @param aggregations TAggregationType.SUM,TAggregationType.COUNT
* @param startTime ()
* @param endTime
* @param interval
* @return SessionDataSet
*/
@Override
public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb聚合查询paths:[{}], aggregations[{}],startTime[{}], endTime:[{}] ,interval:[{}]", paths, aggregations, startTime, endTime, interval);
sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations[{}] ,startTime[{}], endTime:[{}], interval:[{}], error={}", paths, aggregations, startTime, endTime, interval, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* ()
*
* @param paths root.ln.wf01.wt01.temperature
* @param aggregations TAggregationType.SUM,TAggregationType.COUNT
* @param startTime ()
* @param endTime
* @param interval
* @param slidingStep
* @return SessionDataSet
*/
@Override
public SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations, long startTime, long endTime, long interval, long slidingStep) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb聚合查询paths:[{}], aggregations[{}],startTime[{}], endTime:[{}] ,interval:[{}], slidingStep:[{}]", paths, aggregations, startTime, endTime, interval, slidingStep);
sessionDataSetWrapper = sessionPool.executeAggregationQuery(paths, aggregations, startTime, endTime, interval, slidingStep);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeAggregationQuery失败: paths:[{}], aggregations[{}] ,startTime[{}], endTime:[{}], interval:[{}], slidingStep:[{}] ,error={}", paths, aggregations, startTime, endTime, interval, slidingStep, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* SQL
*
* @param sql
* @return
*/
@Override
public SessionDataSet executeQueryStatement(String sql) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
SessionDataSetWrapper sessionDataSetWrapper = null;
try {
log.info("iotdb SQL查询sql:[{}]", sql);
sessionDataSetWrapper = sessionPool.executeQueryStatement(sql);
return sessionDataSetWrapper.getSessionDataSet();
} catch (Exception e) {
log.error("IotDBSession executeQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
} finally {
sessionPool.closeResultSet(sessionDataSetWrapper);
}
return null;
}
/**
* SQL
*
* @param sql
*/
@Override
public void executeNonQueryStatement(String sql) {
SessionPool sessionPool = iotDBSessionConfig.getSessionPool();
try {
log.info("iotdb SQL无查询sql:[{}]", sql);
sessionPool.executeNonQueryStatement(sql);
} catch (Exception e) {
log.error("IotDBSession executeNonQueryStatement失败:sql:[{}],error={}", sql, e.getMessage());
}
}
/**
*
*
* @param sessionDataSet
* @param titleList
*/
@SneakyThrows
@Override
public List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> titleList) {
int fetchSize = sessionDataSet.getFetchSize();
List<Map<String, Object>> resultList = new ArrayList<>();
titleList.remove("Time");
if (fetchSize > 0) {
while (sessionDataSet.hasNext()) {
Map<String, Object> resultMap = new HashMap<>();
RowRecord next = sessionDataSet.next();
List<Field> fields = next.getFields();
String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
resultMap.put("time", timeString);
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
resultMap.put(splitString(titleList.get(i)), null);
} else {
resultMap.put(splitString(titleList.get(i)), field.getObjectValue(field.getDataType()).toString());
}
}
resultList.add(resultMap);
}
}
return resultList;
}
/**
*
*
* @param sessionDataSet
* @param titleList
* @param clazz
* @param <T>
* @return
*/
@SneakyThrows
@Override
public <T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList, Class<? extends IotDbRecordAble> clazz) {
int fetchSize = sessionDataSet.getFetchSize();
List<T> resultList = new ArrayList<>();
titleList.remove("Time");
if (fetchSize > 0) {
while (sessionDataSet.hasNext()) {
Map<String, Object> resultMap = new HashMap<>();
RowRecord next = sessionDataSet.next();
List<Field> fields = next.getFields();
String timeString = new java.text.SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(next.getTimestamp());
resultMap.put("time", timeString);
if (titleList.stream().anyMatch(str -> str.contains("."))) {
for (int i = 0; i < fields.size(); i++) {
Field field = fields.get(i);
String title = titleList.get(i);
if (field.getDataType() == null || field.getObjectValue(field.getDataType()) == null) {
resultMap.put(splitString(title), null);
} else {
resultMap.put(splitString(title), field.getObjectValue(field.getDataType()).toString());
}
}
} else {
Field fieldName = fields.get(0);
Field fieldValue = fields.get(1);
Field fieldDataType = fields.get(2);
if (fieldName.getDataType() != null && fieldName.getObjectValue(fieldName.getDataType()) != null) {
String mapKey = fieldName.getObjectValue(fieldName.getDataType()).toString();
Object mapValue = convertStringToType(fieldValue.getObjectValue(fieldValue.getDataType()).toString(), fieldDataType.getObjectValue(fieldDataType.getDataType()).toString());
resultMap.put(splitString(mapKey), mapValue);
}
}
String jsonString = JSON.toJSONString(resultMap);
resultList.add(JSON.parseObject(jsonString, (Type) clazz));
}
}
return resultList;
}
/**
*
*
* @param str
* @return
*/
public static String splitString(String str) {
String[] parts = str.split("\\.");
if (parts.length <= 0) {
return str;
} else {
return parts[parts.length - 1];
}
}
/**
*
*
* @param value
* @param typeName
* @return
*/
public static Object convertStringToType(String value, String typeName) {
String type = typeName.toLowerCase();
if (type.isEmpty()) {
return value;
}
if ("boolean".equals(type)) {
return Boolean.parseBoolean(value);
} else if ("double".equals(type)) {
return Double.parseDouble(value);
} else if ("int32".equals(type)) {
return Integer.parseInt(value);
} else if ("int64".equals(type)) {
return Long.parseLong(value);
} else if ("float".equals(type)) {
return Float.parseFloat(value);
} else if ("text".equals(type)) {
return value;
} else {
return value;
}
}
/**
* TSDataType
*
* @param type
* @return TSDataType
*/
public static TSDataType getTsDataTypeByString(String type) {
String typeName = splitString(type).toLowerCase();
if ("boolean".equals(typeName)) {
return TSDataType.BOOLEAN;
} else if ("double".equals(typeName)) {
return TSDataType.DOUBLE;
} else if ("int".equals(typeName) || "integer".equals(typeName)) {
return TSDataType.INT32;
} else if ("long".equals(typeName)) {
return TSDataType.INT64;
} else if ("float".equals(typeName)) {
return TSDataType.FLOAT;
} else if ("text".equals(typeName)) {
return TSDataType.TEXT;
} else if ("string".equals(typeName)) {
return TSDataType.TEXT;
} else {
return TSDataType.UNKNOWN;
}
}
/**
* MeasurementSchemas
*
* @param obj
* @return
*/
@Override
public List<MeasurementSchema> buildMeasurementSchemas(Object obj) {
java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
List<MeasurementSchema> schemaList = Arrays.stream(fields).map(field ->
new MeasurementSchema(field.getName(),
getTsDataTypeByString(
field.getType().getName()
))).
collect(Collectors.toList());
return schemaList;
}
/**
* MeasurementSchemaValuesDTO
*
* @param obj
* @return
*/
@SneakyThrows
@Override
public MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object obj) {
MeasurementSchemaValuesDTO measurementSchemaValuesDTO = new MeasurementSchemaValuesDTO();
java.lang.reflect.Field[] fields = obj.getClass().getDeclaredFields();
List<MeasurementSchema> schemaList = new ArrayList<>();
List<Object> values = new ArrayList<>();
List<Integer> valuesIsNullIndex = new ArrayList<>();
int valueIndex = 0;
for (java.lang.reflect.Field field : fields) {
MeasurementSchema measurementSchema = new MeasurementSchema(field.getName(), getTsDataTypeByString(field.getType().getName()));
schemaList.add(measurementSchema);
Object value = field.get(obj);
if (value == null) {
valuesIsNullIndex.add(valueIndex);
}
values.add(value);
valueIndex++;
}
measurementSchemaValuesDTO.setSchemaList(schemaList);
measurementSchemaValuesDTO.setValues(values);
return measurementSchemaValuesDTO;
}
}

View File

@ -20,6 +20,8 @@
<module>cloud-common-system</module>
<module>cloud-common-xxl</module>
<module>cloud-common-rabbit</module>
<module>cloud-common-cache</module>
<module>cloud-common-iotdb</module>
</modules>
<artifactId>cloud-common</artifactId>
@ -29,12 +31,5 @@
cloud-common通用模块
</description>
<dependencies>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-server</artifactId>
<version>3.6.3</version>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,37 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterpise</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>enterpise-cache</artifactId>
<properties>
<maven.compiler.source>17</maven.compiler.source>
<maven.compiler.target>17</maven.compiler.target>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
<dependencies>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-cache</artifactId>
<version>3.6.3</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>enterpise-common</artifactId>
<version>3.6.3</version>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,41 @@
package com.muyu.enterpise.cache;
import com.muyu.common.cache.CacheAbsBasic;
import com.muyu.domain.MessageValue;
import org.springframework.stereotype.Component;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/29 20:04
* @
*/
@Component
public class MessageValueCacheService extends CacheAbsBasic<String, MessageValue> {
/**
*
* @return
*/
@Override
public String keyPre() {
return"messageValue:info:";
}
@Override
public String encode(String key) {
return super.encode(key);
}
/**
*
* @param key
* @return
*/
@Override
public String decode(String key) {
return key.replace("messageValue:info:","");
}
}

View File

@ -0,0 +1,32 @@
package com.muyu.enterpise.cache;
import com.muyu.common.cache.CacheAbsBasic;
import com.muyu.domain.SysCar;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/30 11:06
* @
*/
@Component
public class SysCarCacheService extends CacheAbsBasic<String, SysCar> {
@Override
public String keyPre() {
return "sysCar:info:";
}
@Override
public String encode(String key) {
return super.encode(key);
}
@Override
public String decode(String key) {
return super.decode(key);
}
}

View File

@ -0,0 +1,29 @@
package com.muyu.enterpise.cache;
import com.muyu.common.cache.CacheAbsBasic;
import com.muyu.domain.SysCarType;
import org.springframework.stereotype.Component;
/**
* @version 1.0
* @Author xie ya ru
* @Date 2024/9/30 11:18
* @
*/
@Component
public class SysCarTypeCacheService extends CacheAbsBasic<String, SysCarType> {
@Override
public String keyPre() {
return "sysCarType:info:";
}
@Override
public String encode(String key) {
return super.encode(key);
}
@Override
public String decode(String key) {
return super.decode(key);
}
}

View File

@ -0,0 +1,32 @@
package com.muyu.enterpise.cache;
import com.muyu.common.cache.CacheAbsBasic;
import com.muyu.domain.WarnStrategy;
import org.springframework.stereotype.Component;
/**
* @Authoryan
* @Packagecom.muyu.enterpise.cache
* @Projectplus
* @nameWarnStrategyCacjeService
* @Date2024/10/6 19:26
*/
@Component
public class WarnStrategyCacjeService extends CacheAbsBasic<String, WarnStrategy> {
@Override
public String keyPre() {
return "sysCarType:info:";
}
@Override
public String encode(String key) {
return super.encode(key);
}
@Override
public String decode(String key) {
return super.decode(key);
}
}

View File

@ -5,7 +5,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterprise</artifactId>
<artifactId>cloud-modules-enterpise</artifactId>
<version>3.6.3</version>
</parent>

View File

@ -5,7 +5,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterprise</artifactId>
<artifactId>cloud-modules-enterpise</artifactId>
<version>3.6.3</version>
</parent>
@ -19,17 +19,6 @@
</properties>
<dependencies>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common</artifactId>
<version>3.6.3</version>
</dependency>
<dependency>
<groupId>com.muyu</groupId>
<artifactId>enterpise-common</artifactId>
<version>${muyu.version}</version>
</dependency>
<dependency>
<groupId>org.eclipse.paho</groupId>

View File

@ -5,7 +5,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterprise</artifactId>
<artifactId>cloud-modules-enterpise</artifactId>
<version>3.6.3</version>
</parent>

View File

@ -5,7 +5,7 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>com.muyu</groupId>
<artifactId>cloud-modules-enterprise</artifactId>
<artifactId>cloud-modules-enterpise</artifactId>
<version>3.6.3</version>
</parent>

Some files were not shown because too many files have changed in this diff Show More