six-vehicle
/
cloud-server
5
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(): 新增IoTDB配置类、基础业务类

dev.vehicleGateway
xinzirun 2024-09-29 00:55:53 +08:00
parent 38a096ea72
commit 47567ff055
27 changed files with 1534 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
cloud-auth/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
# Spring
spring:
application:

2
cloud-gateway/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
# Spring
spring:

91
cloud-modules/cloud-modules-data-process/pom.xml 100644
View File

@ -0,0 +1,91 @@
<?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</artifactId>
<version>3.6.3</version>
</parent>
<artifactId>cloud-modules-data-process</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>
<description>
cloud-modules-data-process 数据处理
</description>
<dependencies>
<!-- SpringCloud Alibaba Nacos -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
</dependency>
<!-- SpringCloud Alibaba Nacos Config -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
</dependency>
<!-- SpringCloud Alibaba Sentinel -->
<dependency>
<groupId>com.alibaba.cloud</groupId>
<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
</dependency>
<!-- SpringBoot Actuator -->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<!-- Mysql Connector -->
<dependency>
<groupId>com.mysql</groupId>
<artifactId>mysql-connector-j</artifactId>
</dependency>
<!-- MuYu Common DataSource -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-datasource</artifactId>
</dependency>
<!-- MuYu Common DataScope -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-datascope</artifactId>
</dependency>
<!-- MuYu Common Log -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-log</artifactId>
</dependency>
<!-- 接口模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-api-doc</artifactId>
</dependency>
<!-- 公共核心模块 -->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-core</artifactId>
</dependency>
<!-- IotDB会话 -->
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
</dependency>
</dependencies>
</project>

20
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/CloudDataProcessApplication.java 100644
View File

@ -0,0 +1,20 @@
package com.muyu.data.process;
import com.muyu.common.security.annotation.EnableCustomConfig;
import com.muyu.common.security.annotation.EnableMyFeignClients;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* @Author: zi run
* @Date 2024/9/28 22:31
* @Description 数据处理微服启动类
*/
@EnableCustomConfig
@EnableMyFeignClients
@SpringBootApplication
public class CloudDataProcessApplication {
public static void main(String[] args) {
SpringApplication.run(CloudDataProcessApplication.class, args);
}
}

72
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/basic/config/IoTDBSessionConfig.java 100644
View File

@ -0,0 +1,72 @@
package com.muyu.data.process.basic.config;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.session.pool.SessionPool;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.List;
/**
* @Author: zi run
* @Date 2024/9/28 22:41
* @Description IoTDB会话配置
*/
@Slf4j
@Component
@Configuration
public class IoTDBSessionConfig {
@Value("${spring.iotdb.username}")
private String username;
@Value("${spring.iotdb.password}")
private String password;
@Value("${spring.iotdb.ip}")
private String ip;
@Value("${spring.iotdb.port}")
private int port;
@Value("${spring.iotdb.maxSize}")
private int maxSize;
/**
* IoTDB会话池
*/
private static SessionPool sessionPool = null;
/**
* 获取IoTDB会话对象
* @return ioTDB会话对象
*/
public SessionPool getSessionPool() {
if (sessionPool == null) {
sessionPool = new SessionPool(ip, port, username, password, maxSize);
}
return sessionPool;
}
/**
* 向IoTDB中插入特定设备的记录
*
* @param deviceId 设备的唯一标识符
* @param time 记录的时间戳，以毫秒为单位
* @param measurements 与记录关联的测量名称列表
* @param values 每个测量对应的值列表。值的顺序必须与测量名称一一对应
*
* <p>该方法从会话池中获取一个会话，并尝试将指定的记录插入到 IoTDB 中。
* 如果插入失败，将记录错误信息，便于后续排查。</p>
*/
public void insertRecord(String deviceId, long time, List<String> measurements, List<String> values) {
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());
}
}
}

110
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/basic/service/IoTDBService.java 100644
View File

@ -0,0 +1,110 @@
package com.muyu.data.process.basic.service;
import com.muyu.data.process.domain.dto.IoTDbRecordAble;
import com.muyu.data.process.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: zi run
* @Date 2024/9/28 23:37
* @Description IoTDB业务层
*/
public interface IoTDBService {
/**
* 插入一个 Tablet 对象到 IoTDB 数据库
*
* @param tablet 要插入的 Tablet 对象，包含待写入的数据
*/
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 columnNames
*/
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);
}

714
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/basic/service/impl/IoTDBServiceImpl.java 100644
View File

@ -0,0 +1,714 @@
package com.muyu.data.process.basic.service.impl;
import com.alibaba.fastjson.JSON;
import com.muyu.data.process.basic.config.IoTDBSessionConfig;
import com.muyu.data.process.basic.service.IoTDBService;
import com.muyu.data.process.domain.dto.IoTDbRecordAble;
import com.muyu.data.process.domain.dto.MeasurementSchemaValuesDTO;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.common.rpc.thrift.TAggregationType;
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.isession.SessionDataSet;
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: zi run
* @Date 2024/9/28 23:38
* @Description IoTDB业务实现层
*/
@Slf4j
@Service
@RequiredArgsConstructor
public class IoTDBServiceImpl implements IoTDBService {
/**
* IoTDB会话配置
*/
private final IoTDBSessionConfig ioTDBSessionConfig;
/**
* 插入一个 Tablet 对象到 IoTDB 数据库
*
* @param tablet 要插入的 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());
}
}
/**
* 将给定的 Tablets 插入到 IoTDB 数据库中。
*
* @param tablets 一个 Map，包含要插入的 Tablets，键为 String 类型，值为 Tablet 对象。
* Tablets 应该已经准备好并符合 IoTDB 的插入要求。
*/
@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 temperature，status(字段名)
* @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 temperature，status(字段名)
* @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;
}
}

33
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/DataJSON.java 100644
View File

@ -0,0 +1,33 @@
package com.muyu.data.process.domain;
import io.swagger.v3.oas.annotations.media.Schema;
import io.swagger.v3.oas.annotations.tags.Tag;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
/**
* @Author: zi run
* @Date 2024/9/29 0:11
* @Description JSON数据对象
*/
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
@Tag(name = "ionDB数据源对象")
public class DataJSON {
/**
* 时间戳
*/
@Schema(name = "时间戳")
private Long timestamp;
/**
* 车辆JSON数据
*/
@Schema(name = "车辆JSON数据")
private String datasource;
}

67
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/InsertDataDTO.java 100644
View File

@ -0,0 +1,67 @@
package com.muyu.data.process.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.util.ArrayList;
import java.util.List;
/**
* @Author: zi run
* @Date 2024/9/29 0:12
* @Description 插入数据 数据转换对象
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class InsertDataDTO {
/**
* 温度值
*/
private Float temperature;
/**
* 硬件标识
*/
private String hardware;
/**
* 状态标识
*/
private Boolean status;
/**
* 创建一个单一的InsertDataDTO实例，并设置默认值。
*
* @return 一个配置好的InsertDataDTO对象
*/
public InsertDataDTO buildOne() {
InsertDataDTO insertDataDTO = new InsertDataDTO();
insertDataDTO.setHardware("ss");
insertDataDTO.setStatus(true);
insertDataDTO.setTemperature(12.0F);
return insertDataDTO;
}
/**
* 创建一个单一的InsertDataDTO实例，并设置默认值。
*
* @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;
}
}

36
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/ResultEntity.java 100644
View File

@ -0,0 +1,36 @@
package com.muyu.data.process.domain;
import com.muyu.data.process.domain.dto.IoTDbRecordAble;
import lombok.Data;
import lombok.EqualsAndHashCode;
/**
* @Author: zi run
* @Date 2024/9/29 0:18
* @Description 结果实体
*/
@Data
@EqualsAndHashCode(callSuper = true)
public class ResultEntity extends IoTDbRecordAble {
/**
* 温度值
*/
private Float temperature;
/**
* 硬件标识
*/
private String hardware;
/**
* 状态标识
*/
private Boolean status;
/**
* 时间
*/
private String time;
}

43
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/TestDataType.java 100644
View File

@ -0,0 +1,43 @@
package com.muyu.data.process.domain;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
/**
* @Author: zi run
* @Date 2024/9/29 0:22
* @Description 测试数据类型
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class TestDataType {
/**
* 温度值
*/
private Float temperature;
/**
* 硬件标识
*/
private String hardware;
/**
* 状态标识
*/
private Boolean status;
/**
* 测试Double类型
*/
private Double testDouble;
/**
* 测试Long类型
*/
private Long testLong;
}

12
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/dto/IoTDbRecordAble.java 100644
View File

@ -0,0 +1,12 @@
package com.muyu.data.process.domain.dto;
import lombok.Data;
/**
* @Author: zi run
* @Date 2024/9/29 0:23
* @Description IoTDB数据库记录对象
*/
@Data
public class IoTDbRecordAble {
}

36
cloud-modules/cloud-modules-data-process/src/main/java/com/muyu/data/process/domain/dto/MeasurementSchemaValuesDTO.java 100644
View File

@ -0,0 +1,36 @@
package com.muyu.data.process.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;
/**
* @Author: zi run
* @Date 2024/9/29 0:26
* @Description 测量模式及其对应的值 数据传输对象
*/
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class MeasurementSchemaValuesDTO {
/**
* 测量模式列表，每个元素表示一个测量的定义，包括名称、数据类型等信息。
*/
private List<MeasurementSchema> schemaList;
/**
* 对应于测量模式的实际值列表，存储与 schemaList 中每个测量相对应的值。
*/
private List<Object> values;
/**
* 存储值为空的索引列表
*/
private List<Integer> valueIsNullIndex;
}

2
cloud-modules/cloud-modules-data-process/src/main/resources/banner.txt 100644
View File

@ -0,0 +1,2 @@
Spring Boot Version: ${spring-boot.version}
Spring Application Name: ${spring.application.name}

45
cloud-modules/cloud-modules-data-process/src/main/resources/bootstrap.yml 100644
View File

@ -0,0 +1,45 @@
# Tomcat
server:
port: 11000
nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: xzr
spring:
application:
# 应用名称
name: cloud-data-process
profiles:
# 环境配置
active: dev
cloud:
nacos:
discovery:
# 服务注册地址
server-addr: ${nacos.addr}
# nacos用户名
username: ${nacos.user-name}
# nacos密码
password: ${nacos.password}
# 命名空间
namespace: ${nacos.namespace}
config:
# 服务注册地址
server-addr: ${nacos.addr}
# nacos用户名
username: ${nacos.user-name}
# nacos密码
password: ${nacos.password}
# 命名空间
namespace: ${nacos.namespace}
# 配置文件格式
file-extension: yml
# 共享配置
shared-configs:
# 系统共享配置
- application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
# 系统环境Config共享配置
- application-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}

74
cloud-modules/cloud-modules-data-process/src/main/resources/logback/dev.xml 100644
View File

@ -0,0 +1,74 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
<!-- 日志存放路径 -->
<property name="log.path" value="logs/cloud-data-process"/>
<!-- 日志输出格式 -->
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
<!-- 控制台输出 -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${log.pattern}</pattern>
</encoder>
</appender>
<!-- 系统日志输出 -->
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/info.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<encoder>
<pattern>${log.pattern}</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>INFO</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/error.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<encoder>
<pattern>${log.pattern}</pattern>
</encoder>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>ERROR</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!-- 系统模块日志级别控制 -->
<logger name="com.muyu" level="info"/>
<!-- Spring日志级别控制 -->
<logger name="org.springframework" level="warn"/>
<root level="info">
<appender-ref ref="console"/>
</root>
<!--系统操作日志-->
<root level="info">
<appender-ref ref="file_info"/>
<appender-ref ref="file_error"/>
</root>
</configuration>

81
cloud-modules/cloud-modules-data-process/src/main/resources/logback/prod.xml 100644
View File

@ -0,0 +1,81 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
<!-- 日志存放路径 -->
<property name="log.path" value="logs/cloud-data-process"/>
<!-- 日志输出格式 -->
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
<property name="log.sky.pattern" value="%d{HH:mm:ss.SSS} %yellow([%tid]) [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
<!-- 控制台输出 -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${log.sky.pattern}</pattern>
</encoder>
</appender>
<!-- 系统日志输出 -->
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/info.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>INFO</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/error.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>ERROR</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!-- 使用gRpc将日志发送到skywalking服务端 -->
<appender name="GRPC_LOG" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
<layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternLogbackLayout">
<Pattern>${log.sky.pattern}</Pattern>
</layout>
</encoder>
</appender>
<!-- 系统模块日志级别控制 -->
<logger name="com.muyu" level="info"/>
<!-- Spring日志级别控制 -->
<logger name="org.springframework" level="warn"/>
<root level="info">
<appender-ref ref="GRPC_LOG"/>
<appender-ref ref="console"/>
</root>
<!--系统操作日志-->
<root level="info">
<appender-ref ref="file_info"/>
<appender-ref ref="file_error"/>
</root>
</configuration>

81
cloud-modules/cloud-modules-data-process/src/main/resources/logback/test.xml 100644
View File

@ -0,0 +1,81 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60 seconds" debug="false">
<!-- 日志存放路径 -->
<property name="log.path" value="logs/cloud-data-process"/>
<!-- 日志输出格式 -->
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
<property name="log.sky.pattern" value="%d{HH:mm:ss.SSS} %yellow([%tid]) [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
<!-- 控制台输出 -->
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>${log.sky.pattern}</pattern>
</encoder>
</appender>
<!-- 系统日志输出 -->
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/info.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>INFO</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.path}/error.log</file>
<!-- 循环政策：基于时间创建日志文件 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志文件名格式 -->
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
<!-- 日志最大的历史 60天 -->
<maxHistory>60</maxHistory>
</rollingPolicy>
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<!-- 过滤的级别 -->
<level>ERROR</level>
<!-- 匹配时的操作：接收（记录） -->
<onMatch>ACCEPT</onMatch>
<!-- 不匹配时的操作：拒绝（不记录） -->
<onMismatch>DENY</onMismatch>
</filter>
</appender>
<!-- 使用gRpc将日志发送到skywalking服务端 -->
<appender name="GRPC_LOG" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
<layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternLogbackLayout">
<Pattern>${log.sky.pattern}</Pattern>
</layout>
</encoder>
</appender>
<!-- 系统模块日志级别控制 -->
<logger name="com.muyu" level="info"/>
<!-- Spring日志级别控制 -->
<logger name="org.springframework" level="warn"/>
<root level="info">
<appender-ref ref="GRPC_LOG"/>
<appender-ref ref="console"/>
</root>
<!--系统操作日志-->
<root level="info">
<appender-ref ref="file_info"/>
<appender-ref ref="file_error"/>
</root>
</configuration>

1
cloud-modules/cloud-modules-enterprise/cloud-modules-enterprise-server/pom.xml
View File

@ -22,7 +22,6 @@
</description>
<dependencies>
<!-- SpringCloud Alibaba Nacos -->
<dependency>
<groupId>com.alibaba.cloud</groupId>

2
cloud-modules/cloud-modules-enterprise/cloud-modules-enterprise-server/src/main/java/com/muyu/enterprise/CloudEnterpriseApplication.java
View File

@ -6,7 +6,7 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
/**
* 系统模块
* 企业平台微服务启动类
*
* @author muyu
*/

2
cloud-modules/cloud-modules-enterprise/cloud-modules-enterprise-server/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
spring:
application:

2
cloud-modules/cloud-modules-file/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
# Spring
spring:

2
cloud-modules/cloud-modules-gen/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
# Spring
spring:

2
cloud-modules/cloud-modules-system/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
spring:
application:

1
cloud-modules/pom.xml
View File

@ -14,6 +14,7 @@
<module>cloud-modules-file</module>
<module>cloud-modules-enterprise</module>
<module>cloud-weixin-mp</module>
<module>cloud-modules-data-process</module>
</modules>
<artifactId>cloud-modules</artifactId>

2
cloud-visual/cloud-visual-monitor/src/main/resources/bootstrap.yml
View File

@ -6,7 +6,7 @@ nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
namespace: xzr
# Spring
spring:

8
pom.xml
View File

@ -44,6 +44,7 @@
<xxl-job-core.version>2.4.1</xxl-job-core.version>
<swagger.an.jakarta.verison>2.2.8</swagger.an.jakarta.verison>
<kafka.clients.verison>3.0.0</kafka.clients.verison>
<iotdb-session.verison>1.3.1</iotdb-session.verison>
</properties>
<!-- 依赖声明 -->
@ -199,6 +200,13 @@
<version>${kafka.clients.verison}</version>
</dependency>
<!-- IotDB会话-->
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</artifactId>
<version>${iotdb-session.verison}</version>
</dependency>
<!-- 核心模块 -->
<dependency>
<groupId>com.muyu</groupId>