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

Merge branch 'dev.eventProcessing' into dev 

# Conflicts:
#	cloud-modules/cloud-modules-enterprise/cloud-modules-enterprise-server/pom.xml
#	cloud-modules/pom.xml
dev.vehicleGateway
xinzirun 2024-09-30 18:10:51 +08:00
parent 497de4ade9 5203c69a90
commit c67d648a8a
33 changed files with 2294 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -142,4 +142,4 @@
</plugins>
</build>
</project>
</project>

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
*/

97
cloud-modules/cloud-modules-event-process/pom.xml 100644
View File

@ -0,0 +1,97 @@
<?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-event-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-event-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>
<!-- IotDB会话 -->
<dependency>
<groupId>org.apache.iotdb</groupId>
<artifactId>iotdb-session</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>
<!-- kafka模块-->
<dependency>
<groupId>com.muyu</groupId>
<artifactId>cloud-common-kafka</artifactId>
</dependency>
</dependencies>
</project>

20
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/CloudEventProcessApplication.java 100644
View File

@ -0,0 +1,20 @@
package com.muyu.event.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 CloudEventProcessApplication {
public static void main(String[] args) {
SpringApplication.run(CloudEventProcessApplication.class, args);
}
}

37
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEvent.java 100644
View File

@ -0,0 +1,37 @@
package com.muyu.event.process.basic;
import org.springframework.context.ApplicationEvent;
/**
* @Author: zi run
* @Date 2024/9/30 15:11
* @Description 基础事件
*/
public class BasicEvent<T> extends ApplicationEvent {
/**
* 事件携带的数据
*/
private final T data;
/**
* 构造函数，初始化事件源和数据
*
* @param source 事件源对象
* @param data 事件携带的数据
*/
public BasicEvent(Object source, T data) {
super(source);
this.data = data;
}
/**
* 获取事件携带的数据
*
* @return 事件数据
*/
public T getData() {
return data;
}
}

37
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEventHandler.java 100644
View File

@ -0,0 +1,37 @@
package com.muyu.event.process.basic;
import org.springframework.context.ApplicationListener;
import org.springframework.stereotype.Component;
/**
* @Author: zi run
* @Date 2024/9/30 15:37
* @Description 基础事件处理器
*/
@Component
public class BasicEventHandler<T> implements ApplicationListener<BasicEvent<T>> {
/**
* 具体事件监听器
*/
private final BasicEventListener<T> listener;
/**
* 构造函数，用于注入具体事件监听器
*
* @param listener 具体事件监听器
*/
public BasicEventHandler(BasicEventListener<T> listener) {
this.listener = listener;
}
/**
* 处理应用事件
*
* @param event 事件对象
*/
@Override
public void onApplicationEvent(BasicEvent<T> event) {
listener.onEvent(event);
}
}

16
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEventListener.java 100644
View File

@ -0,0 +1,16 @@
package com.muyu.event.process.basic;
/**
* @Author: zi run
* @Date 2024/9/30 15:35
* @Description 基础事件监听器
*/
public interface BasicEventListener<T> {
/**
* 处理事件的方法
*
* @param event 事件对象
*/
void onEvent(BasicEvent<T> event);
}

38
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/EventPublisher.java 100644
View File

@ -0,0 +1,38 @@
package com.muyu.event.process.basic;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ApplicationEventPublisherAware;
import org.springframework.stereotype.Component;
/**
* @Author: zi run
* @Date 2024/9/29 22:01
* @Description 事件发布者
*/
@Component
public class EventPublisher implements ApplicationEventPublisherAware {
/**
* 应用程序事件发布者，用于发布事件
*/
private ApplicationEventPublisher publisher;
/**
* 设置应用程序事件发布者
*
* @param publisher 应用程序事件发布者实例
*/
@Override
public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
this.publisher = publisher;
}
/**
* 发布事件
* @param event 要发布的事件
* @param <T> 事件数据类型
*/
public <T> void publish(BasicEvent<T> event) {
publisher.publishEvent(event);
}
}

52
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/consumer/TestConsumer.java 100644
View File

@ -0,0 +1,52 @@
package com.muyu.event.process.consumer;
import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Collection;
/**
* @Author: zi run
* @Date 2024/9/29 16:53
* @Description 测试消费者
*/
@Slf4j
//@Component
@RequiredArgsConstructor
public class TestConsumer implements InitializingBean {
/**
* kafka消费者
*/
private final KafkaConsumer<String, String> kafkaConsumer;
/**
* kafka主题名称
*/
private static final String topicName = "test-topic";
@Override
public void afterPropertiesSet() throws Exception {
new Thread(() -> {
log.info("启动线程监听Topic: {}", topicName);
ThreadUtil.sleep(1000);
Collection<String> topics = Lists.newArrayList(topicName);
kafkaConsumer.subscribe(topics);
while (true) {
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
consumerRecords.forEach(record -> {
String value = record.value();
log.info("从Kafka中消费的原始数据: {}", value);
});
}
}).start();
}
}

93
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/consumer/VehicleConsumer.java 100644
View File

@ -0,0 +1,93 @@
package com.muyu.event.process.consumer;
import com.muyu.event.process.basic.EventPublisher;
import com.muyu.event.process.event.IoTDBInsertDataEvent;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextClosedEvent;
import org.springframework.stereotype.Component;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
/**
* @Author: zi run
* @Date 2024/9/29 23:23
* @Description 车辆消费者
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class VehicleConsumer implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {
/**
* kafka消费者
*/
private final KafkaConsumer<String, String> kafkaConsumer;
/**
* 事件发布者
*/
private final EventPublisher eventPublisher;
/**
* 协议解析报文传递数据(队列名称)
*/
public final static String MESSAGE_PARSING = "MessageParsing";
/**
* 设定固定大小的线程池，线程数量与当前可用的处理器核心数相同
*/
private final ExecutorService executorService =
Executors.newFixedThreadPool(10);
@Override
public void run(ApplicationArguments args) throws Exception {
log.info("启动线程监听Topic: {}", MESSAGE_PARSING);
List<String> topics = Collections.singletonList(MESSAGE_PARSING);
kafkaConsumer.subscribe(topics);
while (true) {
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
consumerRecords.forEach(consumerRecord -> executorService.submit(() -> handleRecord(consumerRecord)));
}
}
private void handleRecord(ConsumerRecord<String, String> consumerRecord) {
String message = consumerRecord.value();
log.info("接收到车辆报文数据，内容：{}", message);
log.info("------------------------------------------------");
eventPublisher.publish(new IoTDBInsertDataEvent(this, message));
}
@Override
public void onApplicationEvent(ContextClosedEvent event) {
log.info("关闭线程池和Kafka消费者");
try {
executorService.shutdown();
if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
executorService.shutdownNow();
}
} catch (InterruptedException e) {
log.error("线程池关闭被中断，强制关闭", e);
executorService.shutdownNow();
Thread.currentThread().interrupt();
}
try {
kafkaConsumer.close(); // 关闭Kafka消费者
} catch (Exception e) {
log.error("关闭Kafka消费者时发生错误", e);
}
}
}

88
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/controller/TestEventController.java 100644
View File

@ -0,0 +1,88 @@
package com.muyu.event.process.controller;
import com.alibaba.fastjson2.JSONObject;
import com.muyu.common.core.constant.Constants;
import com.muyu.common.core.domain.Result;
import com.muyu.common.core.web.controller.BaseController;
import com.muyu.event.process.iotdb.service.TestIoTDBService;
import lombok.RequiredArgsConstructor;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
/**
* @Author: zi run
* @Date 2024/9/29 16:24
* @Description 测试事件控制层
*/
@RestController
@RequiredArgsConstructor
@RequestMapping(value = "/test-event")
public class TestEventController extends BaseController {
/**
* kafka生产者
*/
private final KafkaProducer<String, String> kafkaProducer;
/**
* kafka主题名称
*/
private static final String kafkaTopicName = "test-topic";
/**
* 测试IoTDB业务层
*/
private final TestIoTDBService testIoTDBService;
/**
* 发送Kafka测试消息
*
* @return 响应结果
*/
@GetMapping(value = "/sendKafka")
public Result<String> senKafka() {
JSONObject jsonObject = new JSONObject();
jsonObject.put("id","1");
jsonObject.put("name","张三");
jsonObject.put("age","18");
jsonObject.put("sex","男");
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(kafkaTopicName, jsonObject.toJSONString());
kafkaProducer.send(producerRecord);
return Result.success(null, Constants.SUCCESS_MESSAGE);
}
/**
* 查询IoTDB数据列表
* @return 响应结果
*/
@GetMapping(value = "/list")
public Result<List<Map<String, Object>>> list() {
return Result.success(testIoTDBService.list(), Constants.SUCCESS_MESSAGE);
}
/**
* 向IoTDB添加数据
*
* @return 响应结果
*/
@PostMapping(value = "/save")
public Result<String> save() {
String deviceId = "root.test";
ArrayList<String> keyList = new ArrayList<>();
ArrayList<String> valueList = new ArrayList<>();
keyList.add("car_vin");
keyList.add("car_name");
valueList.add("VIN123456");
valueList.add("宝马");
testIoTDBService.insertStringRecord(deviceId, System.currentTimeMillis(), keyList, valueList);
return Result.success(null, Constants.SUCCESS_MESSAGE);
}
}

20
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/event/IoTDBInsertDataEvent.java 100644
View File

@ -0,0 +1,20 @@
package com.muyu.event.process.event;
import com.muyu.event.process.basic.BasicEvent;
/**
* @Author: zi run
* @Date 2024/9/29 21:19
* @Description 向IoTDB插入数据事件
*/
public class IoTDBInsertDataEvent extends BasicEvent<String> {
/**
* 构造函数，向IoTDB插入数据创建事件
*
* @param messsge 消息
*/
public IoTDBInsertDataEvent(Object source, String messsge) {
super(source, messsge);
}
}

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

@ -0,0 +1,72 @@
package com.muyu.event.process.iotdb.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());
}
}
}

290
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/basic/service/IService.java 100644
View File

@ -0,0 +1,290 @@
package com.muyu.event.process.iotdb.basic.service;
import com.muyu.event.process.iotdb.domain.dto.IoTDbRecordAble;
import com.muyu.event.process.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: zi run
* @Date 2024/9/28 23:37
* @Description IoTDB基准业务层
*/
public interface IService {
/**
* 插入一个 Tablet 对象到 IoTDB 数据库
*
* @param tablet 要插入的 Tablet 对象，包含待写入的数据
*/
void insertTablet(Tablet tablet);
/**
* 将给定的 Tablets 插入到 IoTDB 数据库中。
*
* @param tablets 一个 Map，包含要插入的 Tablets
*/
void insertTablets(Map<String, Tablet> tablets);
/**
* 单条数据插入（string类型数据项）
*
* @param deviceId 设备名（表名）root.ln.wf01.wt01
* @param time 时间戳
* @param measurements 数据项列表
* @param values 数据项对应值列表
*/
void insertStringRecord(String deviceId, long time, List<String> measurements, List<String> values);
/**
* 单条数据插入（不同类型数据项）
*
* @param deviceId 设备名（表名）root.ln.wf01.wt01
* @param time 时间戳
* @param measurements 数据项列表
* @param types 数据项对应类型列表
* @param values 数据项对应值列表
*/
void insertRecord(String deviceId, long time, List<String> measurements,
List<TSDataType> types, List<Object> values);
/**
* 多个设备多条数据插入（string类型数据项）
*
* @param deviceIds 多个设备名（表名）root.ln.wf01.wt01
* @param times 时间戳的列表
* @param measurementsList 数据项列表的列表
* @param valuesList 数据项对应值列表的列表
*/
void insertStringRecords(List<String> deviceIds, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList);
/**
* 多个设备多条数据插入（不同类型数据项）
*
* @param deviceIds 多个设备名（表名））root.ln.wf01.wt01
* @param times 时间戳的列表
* @param measurementsList 数据项列表的列表
* @param typesList 数据项对应类型列表的列表
* @param valuesList 数据项对应值列表的列表
*/
void insertRecords(List<String> deviceIds, List<Long> times, List<List<String>> measurementsList,
List<List<TSDataType>> typesList, List<List<Object>> valuesList);
/**
* 单个设备多条数据插入（string类型数据项）
*
* @param deviceId 单个设备名（表名））root.ln.wf01.wt01
* @param times 时间戳的列表
* @param measurementsList 数据项列表的列表
* @param valuesList 数据项对应值列表的列表
*/
void insertStringRecordsOfOneDevice(String deviceId, List<Long> times,
List<List<String>> measurementsList, List<List<String>> valuesList);
/**
* 单个设备多条数据插入（不同类型数据项）
*
* @param deviceId 单个设备名（表名））root.ln.wf01.wt01
* @param times 时间戳的列表
* @param measurementsList 数据项列表的列表
* @param typesList 数据项对应类型列表的列表
* @param valuesList 数据项对应值列表的列表
*/
void insertRecordsOfOneDevice(String deviceId, List<Long> times, List<List<String>> measurementsList,
List<List<TSDataType>> typesList, List<List<Object>> valuesList);
/**
* 删除数据（删除一个时间序列在某个时间点前或这个时间点的数据）
*
* @param path 单个字段 root.ln.wf01.wt01.temperature
* @param endTime 删除时间点
*/
void deleteData(String path, long endTime);
/**
* 删除数据（删除多个时间序列在某个时间点前或这个时间点的数据）
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param endTime 删除时间点
*/
void deleteData(List<String> paths, long endTime);
/**
* 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param startTime 开始时间
* @param endTime 结束时间
* @param outTime 超时时间
* @return SessionDataSet (Time,paths)
*/
SessionDataSet executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime);
/**
* 数据查询（时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间）
*
* @param paths 多个字段（表名），例如："root.ln.wf01.wt01.temperature"
* @param startTime 查询数据的起始时间（包含该时间点）
* @param endTime 查询数据的结束时间（不包含该时间点）
* @param outTime 超时时间，单位为毫秒，表示查询的最长等待时间
* @param clazz 返回数据对应的对象类型，要求对象属性与数据库字段名一致
* @param <T> 返回数据的对象类型泛型
* @return 查询结果的对象列表，如果查询失败则返回 null
*/
<T> List<T> executeRawDataQuery(List<String> paths, long startTime, long endTime, long outTime,
Class<? extends IoTDbRecordAble> clazz);
/**
* 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param lastTime 结束时间
* @return SessionDataSet
*/
SessionDataSet executeLastDataQuery(List<String> paths, long lastTime);
/**
* 最新点查询(查询最后一条时间戳大于等于某个时间点的数据)
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param lastTime 结束时间
* @param clazz 返回数据对应的对象（对象属性必须与字段名对应）
* @return 查询结果的对象列表，如果查询失败则返回 null
* @param <T> 返回数据的对象类型泛型
*/
<T> List<T> executeLastDataQuery(List<String> paths, long lastTime, Class<? extends IoTDbRecordAble> clazz);
/**
* 最新点查询（快速查询单设备下指定序列最新点）
*
* @param db root.ln.wf01
* @param device root.ln.wf01.wt01
* @param sensors temperature，status(字段名)
* @param isLegalPathNodes true(避免路径校验)
* @return SessionDataSet
*/
SessionDataSet executeLastDataQueryForOneDevice(String db, String device,
List<String> sensors, boolean isLegalPathNodes);
/**
* 查询单个设备的最新数据（获取指定设备的最新传感器数据）
*
* @param db root.ln.wf01
* @param device root.ln.wf01.wt01
* @param sensors temperature，status(字段名)
* @param isLegalPathNodes true(避免路径校验)
* @param clazz 返回数据对应的对象（对象属性必须与字段名对应）
* @return 查询结果的对象列表，如果查询失败则返回 null
* @param <T> 返回数据的对象类型泛型
*/
<T> List<T> executeLastDataQueryForOneDevice(String db, String device, List<String> sensors,
boolean isLegalPathNodes, Class<? extends IoTDbRecordAble> clazz);
/**
* 聚合查询
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
* @return SessionDataSet
*/
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations);
/**
* 聚合查询(时间序列原始数据范围查询，时间范围为左闭右开区间，包含开始时间但不包含结束时间)
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
* @param startTime 开始时间(包含)
* @param endTime 结束时间
* @return SessionDataSet
*/
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations,
long startTime, long endTime);
/**
* 聚合查询(支持按照时间区间分段查询)
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
* @param startTime 开始时间(包含)
* @param endTime 结束时间
* @param interval 查询的时间间隔（单位为毫秒）
* @return SessionDataSet
*/
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations,
long startTime, long endTime, long interval);
/**
* 聚合查询(支持按照时间区间分段查询)
*
* @param paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param aggregations 聚合操作 TAggregationType.SUM,TAggregationType.COUNT
* @param startTime 开始时间(包含)
* @param endTime 结束时间
* @param interval 查询的时间间隔（单位为毫秒）
* @param slidingStep 滑动步长（单位为毫秒）
* @return SessionDataSet
*/
SessionDataSet executeAggregationQuery(List<String> paths, List<TAggregationType> aggregations,
long startTime, long endTime, long interval, long slidingStep);
/**
* SQL查询
*
* @param sql SQL查询语句，支持IotDB的查询语法
* @return 返回查询结果的 SessionDataSet，如果执行失败则返回 null
*/
SessionDataSet executeQueryStatement(String sql);
/**
* SQL非查询
*
* @param sql SQL查询语句，支持IotDB的查询语法
*/
void executeNonQueryStatement(String sql);
/**
* 封装处理数据
*
* @param sessionDataSet 包含查询结果的SessionDataSet对象
* @param titleList 列标题列表，用于映射字段名称
* @return 返回封装后的数据列表，每个 Map 代表一行数据，键为列名，值为对应的字段值
*/
List<Map<String, Object>> packagingMapData(SessionDataSet sessionDataSet, List<String> titleList);
/**
* 封装处理数据（不支持聚合查询）
*
* @param sessionDataSet 查询返回的结果集
* @param titleList 查询返回的结果集内的字段名
* @param clazz 返回数据对应的对象（对象属性必须与字段名对应）
* @return 返回封装后的对象列表，每个对象对应一行结果集数据
* @param <T> 返回对象的类型
*/
<T> List<T> packagingObjectData(SessionDataSet sessionDataSet, List<String> titleList,
Class<? extends IoTDbRecordAble> clazz);
/**
* 根据对象构建MeasurementSchemas
*
* @param obj 要从中提取字段信息的对象
* @return 返回一个包含 MeasurementSchema 的列表
*/
List<MeasurementSchema> buildMeasurementSchemas(Object obj);
/**
* 根据对象构建MeasurementSchemaValuesDTO
*
* @param obj 要从中提取字段信息和对应值的对象
* @return MeasurementSchemaValuesDTO 对象
*/
MeasurementSchemaValuesDTO buildMeasurementSchemasAndValues(Object obj);
}

765
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/basic/service/impl/ServiceImpl.java 100644
View File

@ -0,0 +1,765 @@
package com.muyu.event.process.iotdb.basic.service.impl;
import com.alibaba.fastjson.JSON;
import com.muyu.event.process.iotdb.basic.config.IoTDBSessionConfig;
import com.muyu.event.process.iotdb.basic.service.IService;
import com.muyu.event.process.iotdb.domain.dto.IoTDbRecordAble;
import com.muyu.event.process.iotdb.domain.dto.MeasurementSchemaValuesDTO;
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.beans.factory.annotation.Autowired;
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
public class ServiceImpl implements IService {
/**
* IoTDB会话配置
*/
@Autowired
private 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
*/
@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 查询结果的对象列表，如果查询失败则返回 null
*/
@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 paths 多个字段（表名）） root.ln.wf01.wt01.temperature
* @param lastTime 结束时间
* @param clazz 返回数据对应的对象（对象属性必须与字段名对应）
* @return 查询结果的对象列表，如果查询失败则返回 null
* @param <T> 返回数据的对象类型泛型
*/
@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 返回数据对应的对象（对象属性必须与字段名对应）
* @return 查询结果的对象列表，如果查询失败则返回 null
* @param <T> 返回数据的对象类型泛型
*/
@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 SQL查询语句，支持IotDB的查询语法
* @return 返回查询结果的 SessionDataSet，如果执行失败则返回 null
*/
@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 SQL查询语句，支持IotDB的查询语法
*/
@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 包含查询结果的SessionDataSet对象
* @param titleList 列标题列表，用于映射字段名称
* @return 返回封装后的数据列表，每个 Map 代表一行数据，键为列名，值为对应的字段值
*/
@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 返回数据对应的对象（对象属性必须与字段名对应）
* @return 返回封装后的对象列表，每个对象对应一行结果集数据
* @param <T> 返回对象的类型
*/
@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 返回一个包含 MeasurementSchema 的列表
*/
@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 MeasurementSchemaValuesDTO 对象
*/
@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-event-process/src/main/java/com/muyu/event/process/iotdb/domain/DataJSON.java 100644
View File

@ -0,0 +1,33 @@
package com.muyu.event.process.iotdb.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;
}

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

@ -0,0 +1,36 @@
package com.muyu.event.process.iotdb.domain;
import com.muyu.event.process.iotdb.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-event-process/src/main/java/com/muyu/event/process/iotdb/domain/TestDataType.java 100644
View File

@ -0,0 +1,43 @@
package com.muyu.event.process.iotdb.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;
}

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

@ -0,0 +1,67 @@
package com.muyu.event.process.iotdb.domain.dto;
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;
}
}

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

@ -0,0 +1,12 @@
package com.muyu.event.process.iotdb.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-event-process/src/main/java/com/muyu/event/process/iotdb/domain/dto/MeasurementSchemaValuesDTO.java 100644
View File

@ -0,0 +1,36 @@
package com.muyu.event.process.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;
/**
* @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;
}

11
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/service/IoTDBService.java 100644
View File

@ -0,0 +1,11 @@
package com.muyu.event.process.iotdb.service;
import com.muyu.event.process.iotdb.basic.service.IService;
/**
* @Author: zi run
* @Date 2024/9/29 22:38
* @Description IoTDB业务层
*/
public interface IoTDBService extends IService {
}

20
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/service/TestIoTDBService.java 100644
View File

@ -0,0 +1,20 @@
package com.muyu.event.process.iotdb.service;
import com.muyu.event.process.iotdb.basic.service.IService;
import java.util.List;
import java.util.Map;
/**
* @Author: zi run
* @Date 2024/9/29 17:23
* @Description 测试IoTDB业务层
*/
public interface TestIoTDBService extends IService {
/**
* 查询IoTDB数据列表
* @return 返回结果
*/
List<Map<String, Object>> list();
}

14
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/service/impl/IoTDBServiceImpl.java 100644
View File

@ -0,0 +1,14 @@
package com.muyu.event.process.iotdb.service.impl;
import com.muyu.event.process.iotdb.basic.service.impl.ServiceImpl;
import com.muyu.event.process.iotdb.service.IoTDBService;
import org.springframework.stereotype.Service;
/**
* @Author: zi run
* @Date 2024/9/29 22:39
* @Description IoTDB业务实现层
*/
@Service
public class IoTDBServiceImpl extends ServiceImpl implements IoTDBService {
}

33
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/iotdb/service/impl/TestIoTDBServiceImpl.java 100644
View File

@ -0,0 +1,33 @@
package com.muyu.event.process.iotdb.service.impl;
import com.muyu.event.process.iotdb.basic.service.impl.ServiceImpl;
import com.muyu.event.process.iotdb.service.TestIoTDBService;
import lombok.extern.slf4j.Slf4j;
import org.apache.iotdb.isession.SessionDataSet;
import org.springframework.stereotype.Service;
import java.util.List;
import java.util.Map;
/**
* @Author: zi run
* @Date 2024/9/29 17:24
* @Description 测试IoTDB业务实现层
*/
@Slf4j
@Service
public class TestIoTDBServiceImpl extends ServiceImpl implements TestIoTDBService {
/**
* 查询IoTDB数据列表
* @return 返回结果
*/
@Override
public List<Map<String, Object>> list() {
String sql = "select * from root.test";
SessionDataSet sessionDataSet = this.executeQueryStatement(sql);
List<Map<String, Object>> list = this.packagingMapData(sessionDataSet, sessionDataSet.getColumnTypes());
log.info("查询IoTDB数据为：{}", list.toString());
return list;
}
}

68
cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/listener/IoTDBInsertDataListener.java 100644
View File

@ -0,0 +1,68 @@
package com.muyu.event.process.listener;
import com.alibaba.fastjson2.JSONObject;
import com.muyu.event.process.basic.BasicEvent;
import com.muyu.event.process.basic.BasicEventListener;
import com.muyu.event.process.event.IoTDBInsertDataEvent;
import com.muyu.event.process.iotdb.service.IoTDBService;
import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.List;
import java.util.stream.Collectors;
/**
* @Author: zi run
* @Date 2024/9/29 22:12
* @Description 向IoTDB插入数据事件监听器
*/
@Component
@RequiredArgsConstructor
public class IoTDBInsertDataListener implements BasicEventListener<String> {
/**
* IoTDB业务层
*/
private final IoTDBService ioTDBService;
/**
* 设备名（表名）
*/
private static final String DEVICE_ID = "root.vehicle";
/**
* 处理接收到的事件，将数据插入到 IoTDB
*
* @param event 接收到的事件，包含需要插入的数据
*/
@Override
public void onEvent(BasicEvent<String> event) {
JSONObject data = JSONObject.parseObject(event.getData());
List<String> keyList = extractKeys(data);
List<String> valueList = extractValues(data);
ioTDBService.insertStringRecord(DEVICE_ID, System.currentTimeMillis(), keyList, valueList);
}
/**
* 从给定的JSONObject中提取所有的键
*
* @param data 要提取键的JSONObject
* @return 键的列表
*/
private List<String> extractKeys(JSONObject data) {
return data.keySet().stream().collect(Collectors.toList());
}
/**
* 从给定的 JSONObject 中提取所有的值，并将其转换为字符串
*
* @param data 要提取值的JSONObject
* @return 值的列表，以字符串形式表示
*/
private List<String> extractValues(JSONObject data) {
return data.values().stream()
.map(Object::toString)
.collect(Collectors.toList());
}
}

2
cloud-modules/cloud-modules-event-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}

47
cloud-modules/cloud-modules-event-process/src/main/resources/bootstrap.yml 100644
View File

@ -0,0 +1,47 @@
# Tomcat
server:
port: 11000
nacos:
addr: 106.15.136.7:8848
user-name: nacos
password: nacos
namespace: dev
spring:
application:
# 应用名称
name: cloud-event-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}
# kafka共享配置
- application-kafka-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}

74
cloud-modules/cloud-modules-event-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-event-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-event-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-event-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-event-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-event-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/pom.xml
View File

@ -16,6 +16,7 @@
<module>cloud-weixin-mp</module>
<module>cloud-modules-protocol-analysis</module>
<module>cloud-modules-vehicle-gateway</module>
<module>cloud-modules-event-process</module>
</modules>
<artifactId>cloud-modules</artifactId>

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>