|
|
@@ -0,0 +1,259 @@
|
|
|
+package io.github.pnoker.gateway.comtool;
|
|
|
+
|
|
|
+import com.alibaba.fastjson.JSONObject;
|
|
|
+import io.github.pnoker.gateway.bizmgr.KprDangyangWaterBizFun;
|
|
|
+import io.github.pnoker.gateway.bizmgr.baseInit.KprBaseInitFun;
|
|
|
+import io.github.pnoker.gateway.utils.HttpUtil;
|
|
|
+import org.influxdb.dto.Point;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import java.math.BigDecimal;
|
|
|
+import java.text.SimpleDateFormat;
|
|
|
+import java.time.*;
|
|
|
+import java.time.format.DateTimeFormatter;
|
|
|
+import java.time.temporal.ChronoUnit;
|
|
|
+import java.util.*;
|
|
|
+import java.util.concurrent.TimeUnit;
|
|
|
+import java.util.stream.Collectors;
|
|
|
+
|
|
|
+/**
|
|
|
+ * @ClassName ThreadTask
|
|
|
+ * @Description: TODO 当阳历史数据定制时段的任务线程
|
|
|
+ * @Author LX
|
|
|
+ * @Date 2024/9/3
|
|
|
+ * @Version V1.0
|
|
|
+ **/
|
|
|
+public class ThreadDinzhiHistoryTask implements Runnable{
|
|
|
+ private static final Logger log = LoggerFactory.getLogger(ThreadDinzhiHistoryTask.class);
|
|
|
+
|
|
|
+ private final String taskId;
|
|
|
+
|
|
|
+ private final Integer collectionFrequency;//采集频率
|
|
|
+
|
|
|
+ private String deviceType;
|
|
|
+
|
|
|
+ private List<Map<String,Object>> deviceList = new ArrayList<>();
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ public ThreadDinzhiHistoryTask(String taskId, Integer collectionFrequency, String deviceType, List<Map<String,Object>> deviceList) {
|
|
|
+ this.taskId = taskId;
|
|
|
+ this.collectionFrequency = collectionFrequency;
|
|
|
+ this.deviceType = deviceType;
|
|
|
+ this.deviceList = deviceList;
|
|
|
+ }
|
|
|
+
|
|
|
+ @Override
|
|
|
+ public void run() {
|
|
|
+ try {
|
|
|
+ // 获取当前时间的整时0分0秒,这里的时主要还是定时器的执行任务时间去控制
|
|
|
+ LocalDateTime now = LocalDateTime.now().withSecond(0);
|
|
|
+
|
|
|
+ // 使用提取的值构造新的LocalDateTime对象
|
|
|
+ LocalDateTime firstDayOfMonthWithCurrentTime = now.minusDays(2);
|
|
|
+
|
|
|
+// days.add(now);//测试用
|
|
|
+
|
|
|
+ List<JSONObject> deviceRealtimeDataList = new ArrayList<>();
|
|
|
+ for (Map<String, Object> map : deviceList) {
|
|
|
+// if(String.valueOf(map.get("deviceCode")).equals("106")) {//测试用
|
|
|
+ JSONObject devicerealtimeDataTotal = null;
|
|
|
+ try {
|
|
|
+ Map<String, String> paramRealtime = new HashMap<>();
|
|
|
+ paramRealtime.put("deviceType", deviceType);
|
|
|
+ paramRealtime.put("deviceCode", String.valueOf(map.get("deviceCode")));
|
|
|
+// paramRealtime.put("pageNo", String.valueOf(pageNo));
|
|
|
+// paramRealtime.put("pageSize", String.valueOf(pageSize));
|
|
|
+ paramRealtime.put("startTime", now.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
|
|
|
+ paramRealtime.put("endTime", firstDayOfMonthWithCurrentTime.format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
|
|
|
+ paramRealtime.put("interval", String.valueOf((collectionFrequency < 60 ? 60 : collectionFrequency)));
|
|
|
+ Map<String, String> headers = new HashMap<>();
|
|
|
+ headers.put("Authorization", "Bearer " + KprDangyangWaterBizFun.dangyangToken);
|
|
|
+ devicerealtimeDataTotal = JSONObject.parseObject(HttpUtil.sendGet(
|
|
|
+ KprDangyangWaterBizFun.hisDataListUrl, paramRealtime, headers));
|
|
|
+// System.out.println(taskId + ",日期:" + paramRealtime.get("startTime") +
|
|
|
+// "----" + paramRealtime.get("endTime") + ";" + paramRealtime.get("deviceCode") + "数据:" + devicerealtimeDataTotal.toJSONString());
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("定制时间段历史任务线程" + taskId + " 调用异常:" + ex.getLocalizedMessage());
|
|
|
+ }
|
|
|
+ if (devicerealtimeDataTotal == null || devicerealtimeDataTotal.getInteger("code") != 0 ||
|
|
|
+ devicerealtimeDataTotal.getJSONArray("data") == null
|
|
|
+ || devicerealtimeDataTotal.getJSONArray("data").isEmpty()) {
|
|
|
+ continue;
|
|
|
+ }
|
|
|
+ try {
|
|
|
+ //TODO 查询数据不为空,插入infulxdb
|
|
|
+ for (int i = 0; i < devicerealtimeDataTotal.getJSONArray("data").size(); i++) {
|
|
|
+ JSONObject deviceRealtimeData = devicerealtimeDataTotal.getJSONArray("data").getJSONObject(i);
|
|
|
+// adjustTime(deviceRealtimeData, collectionFrequency);//整点rand处理
|
|
|
+ deviceRealtimeDataList.add(deviceRealtimeData);
|
|
|
+ }
|
|
|
+ } catch (Exception ex) {
|
|
|
+ log.error("定制时间段历史任务线程" + taskId + " " + String.valueOf(map.get("deviceCode")) + "查询异常:" + ex.getLocalizedMessage() + ";resJson:" + devicerealtimeDataTotal.toJSONString());
|
|
|
+ }
|
|
|
+// }
|
|
|
+// break;
|
|
|
+ }
|
|
|
+ //TODO 开始插入数据库
|
|
|
+ //第三方对应的字段集
|
|
|
+ List<String> params = KprBaseInitFun.getInstance().dangyangParams.get(deviceType);
|
|
|
+ for(JSONObject jsonObject : deviceRealtimeDataList){
|
|
|
+ //deviceType为表名
|
|
|
+ //time为当前数据time的纳秒时间戳,已经做过整点处理了
|
|
|
+ Point pointNanos = createPointFromJson(deviceType,jsonObject,params);
|
|
|
+ if(pointNanos!=null) {
|
|
|
+ KprDangyangWaterBizFun.infulxDbUtil.insert(pointNanos);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ System.out.println(taskId+"定时历史数据同步完毕,数据长度:"+deviceRealtimeDataList.size());
|
|
|
+ }catch(Exception ex){
|
|
|
+ ex.printStackTrace();
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ public static LocalDateTime parseStringToDateTime(String time, String format) {
|
|
|
+ DateTimeFormatter df = DateTimeFormatter.ofPattern(format);
|
|
|
+ return LocalDateTime.parse(time, df);
|
|
|
+ }
|
|
|
+
|
|
|
+ public static String getDateTimeToString(LocalDateTime localDateTime, String format) {
|
|
|
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern(format);
|
|
|
+ return localDateTime.format(formatter);
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+
|
|
|
+ public static Point createPointFromJson(String deviceType, JSONObject jsonObject,List<String> params) {
|
|
|
+ // 获取时间戳
|
|
|
+ long nanoTimestamp = convertToNanoTimestamp(jsonObject.getString("time"));
|
|
|
+ if(nanoTimestamp==0L){
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+ // 创建 Point.Builder 对象
|
|
|
+ Point.Builder pointBuilder = Point.measurement(KprBaseInitFun.getInstance().measurementMap.get(deviceType))
|
|
|
+ .tag("dev_id",jsonObject.getString("deviceCode"))
|
|
|
+ .time(nanoTimestamp, TimeUnit.NANOSECONDS);
|
|
|
+
|
|
|
+ // 遍历 JSON 对象的键值对，并添加到 Point.Builder 中
|
|
|
+ for (Map.Entry<String, Object> entry : jsonObject.entrySet()) {
|
|
|
+ String key = entry.getKey();
|
|
|
+ if (!key.equals("time")&&!key.equals("deviceCode")) {
|
|
|
+ processField(pointBuilder,key, entry.getValue(),params);
|
|
|
+ }
|
|
|
+ }
|
|
|
+ if(!pointBuilder.hasFields()){
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ // 构建 Point 对象
|
|
|
+ try {
|
|
|
+ return pointBuilder.build();
|
|
|
+ }catch(Exception ex){
|
|
|
+
|
|
|
+ }
|
|
|
+ return null;
|
|
|
+ }
|
|
|
+
|
|
|
+ //指定列处理
|
|
|
+ private static void processField(Point.Builder pointBuilder, String key, Object value,List<String> params) {
|
|
|
+ if (value == null) {
|
|
|
+ return; // 跳过空值
|
|
|
+ }
|
|
|
+ //String[]长度如果为0,则说明本系统没有对应字段,则用它本身,反之用第二个元素作为字段名
|
|
|
+ List<String[]> newParams = params.stream()
|
|
|
+ .map(s -> s.split("___"))
|
|
|
+ .collect(Collectors.toList());
|
|
|
+
|
|
|
+ for (String[] clies:newParams){
|
|
|
+ boolean exists = Arrays.asList(clies).contains(key);
|
|
|
+ if(exists){
|
|
|
+ String newKey = clies.length>1?clies[1]:clies[0];
|
|
|
+ if (value instanceof Integer) {
|
|
|
+ pointBuilder.addField(newKey, (Integer) value);
|
|
|
+ } else if (value instanceof Long) {
|
|
|
+ pointBuilder.addField(newKey, (Long) value);
|
|
|
+ } else if (value instanceof Boolean) {
|
|
|
+ pointBuilder.addField(newKey, (Boolean) value);
|
|
|
+ } else if (value instanceof String) {
|
|
|
+ pointBuilder.addField(newKey, (String) value);
|
|
|
+ }else if (value instanceof BigDecimal) {
|
|
|
+ pointBuilder.addField(newKey, ((BigDecimal) value).doubleValue());
|
|
|
+ }else{
|
|
|
+ pointBuilder.addField(newKey, (String.valueOf(value)) );
|
|
|
+ }
|
|
|
+ break;
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
+// switch (key) {
|
|
|
+// case "flow":
|
|
|
+// pointBuilder.addField("flow_cur", ((BigDecimal) value).doubleValue());
|
|
|
+// break;
|
|
|
+// default:
|
|
|
+// if (value instanceof Integer) {
|
|
|
+// pointBuilder.addField(key, (Integer) value);
|
|
|
+// } else if (value instanceof Long) {
|
|
|
+// pointBuilder.addField(key, (Long) value);
|
|
|
+// } else if (value instanceof Boolean) {
|
|
|
+// pointBuilder.addField(key, (Boolean) value);
|
|
|
+// } else if (value instanceof String) {
|
|
|
+// pointBuilder.addField(key, (String) value);
|
|
|
+// }else if (value instanceof BigDecimal) {
|
|
|
+// pointBuilder.addField(key, ((BigDecimal) value).doubleValue());
|
|
|
+// }
|
|
|
+// else {
|
|
|
+//// System.err.println("Unsupported data type for key " + key + ": " + value.getClass().getName());
|
|
|
+// }
|
|
|
+// break;
|
|
|
+// }
|
|
|
+ }
|
|
|
+
|
|
|
+
|
|
|
+ //获取CST 纳秒时间戳
|
|
|
+ private static long convertToNanoTimestamp(String dateTimeString) {
|
|
|
+ // 定义日期时间格式
|
|
|
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+ // 解析字符串为 LocalDateTime 对象
|
|
|
+ LocalDateTime dateTime = LocalDateTime.parse(dateTimeString, formatter);
|
|
|
+ // 指定时区为CST（Asia/Shanghai）
|
|
|
+ ZoneId shanghaiZoneId = ZoneId.of("Asia/Shanghai");
|
|
|
+ // 将LocalDateTime对象转换为ZonedDateTime对象
|
|
|
+ ZonedDateTime zonedDateTime = dateTime.atZone(shanghaiZoneId);
|
|
|
+ // 转换为秒级别的时间戳
|
|
|
+ long epochSecond = dateTime.toEpochSecond(ZoneOffset.UTC);
|
|
|
+ // 将ZonedDateTime对象转换为Instant对象（UTC时间点）
|
|
|
+ Instant instant = zonedDateTime.toInstant();
|
|
|
+ // 计算从1970年1月1日00:00:00 UTC以来的纳秒数
|
|
|
+ long nanosecondsSinceEpoch = ChronoUnit.NANOS.between(Instant.EPOCH, instant);
|
|
|
+ long currentTimeNanos = System.currentTimeMillis() * 1_000_000L;
|
|
|
+ if(nanosecondsSinceEpoch>currentTimeNanos){
|
|
|
+ return 0L;
|
|
|
+ }
|
|
|
+ return nanosecondsSinceEpoch;
|
|
|
+ }
|
|
|
+
|
|
|
+ private static final SimpleDateFormat outputDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
|
|
+ public static void adjustTime(JSONObject deviceRealtimeData, int collectionFrequency) {
|
|
|
+ // 获取原始时间字符串
|
|
|
+ String timeString = deviceRealtimeData.getString("time");
|
|
|
+
|
|
|
+ // 定义日期时间格式
|
|
|
+ DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
|
|
|
+
|
|
|
+ // 解析为 LocalDateTime 对象
|
|
|
+ LocalDateTime originalTime = LocalDateTime.parse(timeString, formatter);
|
|
|
+
|
|
|
+ // 计算最接近的时间间隔
|
|
|
+ long seconds = originalTime.getSecond();
|
|
|
+ long adjustedSeconds = Math.round((double) seconds / collectionFrequency) * collectionFrequency;
|
|
|
+
|
|
|
+ // 创建新的时间
|
|
|
+ LocalDateTime adjustedTime = originalTime.truncatedTo(ChronoUnit.MINUTES).plusSeconds(adjustedSeconds);
|
|
|
+
|
|
|
+ // 更新 JSON 对象中的时间
|
|
|
+ deviceRealtimeData.put("time", adjustedTime.format(formatter));
|
|
|
+ }
|
|
|
+
|
|
|
+}
|
