From 14f2c4fce50c871d84f89d2dca2298e0892b4672 Mon Sep 17 00:00:00 2001
From: zhang <zc857179121@qq.com>
Date: 星期二, 05 五月 2026 08:25:26 +0800
Subject: [PATCH] Merge branch 'rcs_master_1.0' into rcs_master_all
---
component/component-Influxdb/src/main/java/com/zy/component/influxdb/service/InfluxDBService.java | 150 ++++++++++++++++++++++++++++++++++++++++++++++++++
1 files changed, 150 insertions(+), 0 deletions(-)
diff --git a/component/component-Influxdb/src/main/java/com/zy/component/influxdb/service/InfluxDBService.java b/component/component-Influxdb/src/main/java/com/zy/component/influxdb/service/InfluxDBService.java
new file mode 100644
index 0000000..924ee65
--- /dev/null
+++ b/component/component-Influxdb/src/main/java/com/zy/component/influxdb/service/InfluxDBService.java
@@ -0,0 +1,150 @@
+package com.zy.component.influxdb.service;
+
+import com.influxdb.v3.client.InfluxDBClient;
+import com.influxdb.v3.client.Point;
+import com.influxdb.v3.client.PointValues;
+import com.influxdb.v3.client.query.QueryOptions;
+import com.influxdb.v3.client.query.QueryType;
+import com.influxdb.v3.client.write.WritePrecision;
+import com.zy.component.influxdb.domain.BaseMessage;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
+import java.time.Instant;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+
+@Service
+public class InfluxDBService {
+
+ private static final Logger logger = LoggerFactory.getLogger(InfluxDBService.class);
+
+ @Autowired
+ private InfluxDBClient influxDBClient;
+
+ /**
+ * 鍐欏叆鏁版嵁
+ *
+ * @param measurement 琛ㄥ悕
+ * @param tags 鏍囩
+ * @param fields 瀛楁
+ */
+ public void writeData(String measurement, Map<String, String> tags, Map<String, Object> fields,Long timestamp) {
+ Point point = Point.measurement(measurement)
+ .setTags(tags)
+ .setFields(fields)
+ .setTimestamp(timestamp, WritePrecision.MS);
+ try {
+ influxDBClient.writePoint(point);
+ } catch (Exception e) {
+ logger.error("Failed to write data to the database.");
+ e.printStackTrace();
+ }
+ }
+
+ /**
+ * 鏌ヨ鏁版嵁
+ *
+ * @param sql sql璇彞
+ * @return 鏌ヨ缁撴灉鍒楄〃
+ */
+ public List<Map<String, Object>> queryData(String sql) {
+ try {
+ // 鎵ц鏌ヨ
+ Stream<Map<String, Object>> mapStream = influxDBClient.queryRows(sql);
+ List<Map<String, Object>> result = mapStream.collect(Collectors.toList());
+ logger.info("鏌ヨ鏁版嵁锛歿}", result);
+ return result;
+ } catch (Exception e) {
+ logger.error("Failed to query data from the database.");
+ e.printStackTrace();
+ }
+ return null;
+ }
+ /**
+ * 鏌ヨ鏁版嵁
+ *
+ * @param sql sql璇彞
+ * @return 鏌ヨ缁撴灉鍒楄〃
+ */
+ public <T extends BaseMessage> List<T> queryPoints(String sql, Class<T> clazz) {
+ return queryPoints(sql,new HashMap<>(),clazz);
+ }
+ /**
+ * 鏌ヨ鏁版嵁
+ *
+ * @param sql sql璇彞
+ * @return 鏌ヨ缁撴灉鍒楄〃
+ */
+ public <T extends BaseMessage> List<T> queryPoints(String sql,Map<String,Object> queryParams, Class<T> clazz) {
+ try {
+ // 鎵ц鏌ヨ
+ Stream<PointValues> queryPoints = influxDBClient.queryPoints(sql, queryParams);
+ Field[] declaredFields = clazz.getDeclaredFields();
+
+ // 鍒涘缓涓�涓垪琛ㄧ敤浜庡瓨鍌ㄧ粨鏋�
+ List<T> result = queryPoints.map(point -> {
+ T newInstance = null;
+ try {
+ // 浣跨敤鏃犲弬鏋勯�犲嚱鏁板垱寤烘柊瀹炰緥
+ newInstance = clazz.getDeclaredConstructor().newInstance();
+ } catch (NoSuchMethodException | InstantiationException e) {
+ // 濡傛灉娌℃湁鏃犲弬鏋勯�犲嚱鏁版垨鏃犳硶瀹炰緥鍖栵紝鍒欐姏鍑烘湁鎰忎箟鐨勫紓甯�
+ throw new RuntimeException("鏃犳硶鍒涘缓瀵硅薄瀹炰緥锛岃纭繚绫绘湁鍏叡鏃犲弬鏋勯�犲嚱鏁�: " + clazz.getName(), e);
+ } catch (InvocationTargetException e) {
+ throw new RuntimeException(e);
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+
+ String[] tagNames = point.getTagNames();
+ for (String tagName : tagNames) {
+ for (Field declaredField : declaredFields) {
+ if (declaredField.getName().equals(tagName)) {
+ declaredField.setAccessible(true);
+ try {
+ declaredField.set(newInstance, point.getTag(tagName));
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ break;
+ }
+ }
+ }
+ String[] fieldNames = point.getFieldNames();
+ for (String fieldName : fieldNames) {
+ for (Field declaredField : declaredFields) {
+ if (declaredField.getName().equals(fieldName)) {
+ declaredField.setAccessible(true);
+ try {
+ declaredField.set(newInstance, point.getField(fieldName));
+ } catch (IllegalAccessException e) {
+ throw new RuntimeException(e);
+ }
+ break;
+ }
+ }
+ }
+ newInstance.setTimestamp(point.getTimestamp().longValue());
+ return newInstance;
+ }).collect(Collectors.toList());
+ return result;
+ } catch (Exception e) {
+ logger.error("Failed to query data from the database.");
+ e.printStackTrace();
+ }
+ return null;
+ }
+
+
+
+}
--
Gitblit v1.9.1