package com.iailab.module.data.channel.http.collector.ihdb; import com.alibaba.fastjson.JSON; import com.alibaba.fastjson.JSONArray; import com.alibaba.fastjson.JSONObject; import com.google.gson.Gson; import com.iailab.framework.common.constant.CommonConstant; import com.iailab.framework.common.util.http.HttpUtils; import com.iailab.module.data.channel.http.entity.HttpApiEntity; import com.iailab.module.data.channel.http.service.HttpApiService; import com.iailab.module.data.common.enums.DataSourceType; import com.iailab.module.data.common.utils.DateUtils; import com.iailab.module.data.common.utils.HttpRequest; import com.iailab.module.data.common.utils.TagUtils; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.data.redis.core.BoundHashOperations; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; import java.math.BigDecimal; import java.math.RoundingMode; import java.sql.Timestamp; import java.text.ParseException; import java.text.SimpleDateFormat; import java.time.LocalDateTime; import java.util.*; import java.util.concurrent.*; import java.util.function.Function; import java.util.stream.Collectors; /** * iHyperDB采集 * * @author lirm * @Description * @createTime 2024年10月16日 */ @Slf4j @Component public class HttpCollectorForIhd { private static Map apiMap = new HashMap<>(); @Autowired private HttpApiService httpApiService; @Autowired private RedisTemplate redisTemplate; ThreadPoolExecutor threadPool = new ThreadPoolExecutor(18, 36, 30, TimeUnit.SECONDS, new ArrayBlockingQueue(36), new ThreadPoolExecutor.AbortPolicy()); private static final String STA_TRUE = "true"; private static final int GROUP_MAX_COUNT = 300; private static final int MAX_WAIT = 45; private static final String pattern = "yyyyMMddHHmm00"; private static final String IS_SUCCESS = "isSuccess"; private static final String RESP_DATA = "data"; public static final String TIV = "TagIdValue:"; public static final long offset = 60 * 10L; /** * tagName */ private static final String N = "n"; /** * dimension */ private static final String D = "d"; /** * 类型 */ private static final String P = "p"; /** * dataValue */ private static final String V = "v"; /** * dataTime */ private static final String T = "t"; /** * 数据质量G:good,B:bad */ private static final String Q = "q"; private HttpApiEntity getHttpApi(String id) { if (apiMap.containsKey(id)) { return apiMap.get(id); } HttpApiEntity httpApi = httpApiService.info(id); apiMap.put(id, httpApi); return httpApi; } public BigDecimal getTagValue(String sourceId, String tagNo, Integer dimension, String valueType) { BigDecimal value = CommonConstant.BAD_VALUE; HttpApiEntity httpApi = this.getHttpApi(sourceId); StringBuilder tagSb = new StringBuilder(); tagSb.append("["); Map queryParams = new HashMap<>(); queryParams.put(P, valueType); queryParams.put(D, dimension); queryParams.put(N, tagNo); String jsonString = JSON.toJSONString(queryParams); tagSb.append(jsonString); tagSb.append("]"); String currentDate = DateUtils.format(new Date(), pattern); String responseStr = HttpUtils.sendPost(httpApi.getUrl() + "/" + currentDate, tagSb.toString()); JSONObject responseObj = JSON.parseObject(responseStr); if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { JSONArray tagValueList = responseObj.getJSONArray(RESP_DATA); if (!CollectionUtils.isEmpty(tagValueList)) { for (int i = 0; i < tagValueList.size(); i++) { JSONObject item = tagValueList.getJSONObject(i); value = new BigDecimal(item.get(V).toString()); } } } return value; } public Map getLastValues(String sourceId, List tagNames) { Map result = new HashMap<>(); HttpApiEntity httpApi = this.getHttpApi(sourceId); try { if (CollectionUtils.isEmpty(tagNames)) { return result; } List noCacheTagNames = new ArrayList<>();//未缓存的tag for (String tagName : tagNames) { //先查缓存 BoundHashOperations ops = redisTemplate.boundHashOps("IhdTag:" + tagName); if (ops.get(V) != null) { BigDecimal value = new BigDecimal(ops.get(V).toString()); result.put(tagName, value.setScale(3, RoundingMode.HALF_UP)); } else { noCacheTagNames.add(tagName); } } if (CollectionUtils.isEmpty(noCacheTagNames)) { log.info("全部读取缓存"); return result; } log.info("查询未缓存的数据"); Gson gson = new Gson(); String tagSb = gson.toJson(noCacheTagNames); String currentDate = DateUtils.format(new Date(), pattern); String responseStr = ""; responseStr = HttpUtils.sendPost(httpApi.getUrl().replace("getPointdatasAvg", "getPointslast") + "/" + currentDate, tagSb); JSONObject responseObj = JSON.parseObject(responseStr); if (!STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { return result; } JSONArray tagValueList = responseObj.getJSONArray(RESP_DATA); if (CollectionUtils.isEmpty(tagValueList)) { return result; } for (int i = 0; i < tagValueList.size(); i++) { JSONObject item = tagValueList.getJSONObject(i); if (item.get(V) == null) { continue; } //存缓存 BoundHashOperations ops = redisTemplate.boundHashOps("IhdTag:" + item.get(N).toString()); ops.put(V, item.get(V).toString()); //设置过期时间 redisTemplate.expire(item.get(N).toString(), 10, TimeUnit.SECONDS); //把查询到的数据插入结果集 BigDecimal value = new BigDecimal(item.get(V).toString()); result.put(item.get(N).toString(), value.setScale(3, RoundingMode.HALF_UP)); } } catch (Exception ex) { log.info("getCurrentValue异常"); ex.printStackTrace(); throw ex; } return result; } public Map getTagValues(List params, Date collectTime) { Map result = new HashMap<>(); if (CollectionUtils.isEmpty(params)) { return new HashMap<>(); } try { Map> measurePointsCountGroup = new HashMap<>(); int pointListSize = params.size(); int groupCount = pointListSize / GROUP_MAX_COUNT + ((pointListSize % GROUP_MAX_COUNT) > 0 ? 1 : 0); log.info("groupCount=" + groupCount); for (int i = 0; i < groupCount; i++) { int end = (i + 1) * GROUP_MAX_COUNT; if (end > pointListSize) { end = pointListSize; } measurePointsCountGroup.put(i, params.subList(i * GROUP_MAX_COUNT, end)); } log.info("measurePointsCountGroup.size()=" + measurePointsCountGroup.size()); result = new ConcurrentHashMap<>(params.size()); CountDownLatch countDownLatch = new CountDownLatch(measurePointsCountGroup.size()); List finishedGroup = new CopyOnWriteArrayList<>(); List notFinish = new CopyOnWriteArrayList<>(); for (Map.Entry> measurePointsItem : measurePointsCountGroup.entrySet()) { HttpApiEntity httpApi = this.getHttpApi(measurePointsItem.getValue().get(0)[0].toString()); // 并发 Thread.sleep(500); threadPool.submit(new Task(httpApi.getUrl(), httpApi.getCode(), result, measurePointsItem.getValue(), collectTime, countDownLatch, finishedGroup, measurePointsItem.getKey())); // 顺序 //this.getByHtp(result, measurePointsItem.getValue(), collectTime); } countDownLatch.await(MAX_WAIT, TimeUnit.SECONDS); for (Map.Entry itemValue : result.entrySet()) { redisTemplate.opsForValue().set(TIV + itemValue.getKey(), itemValue.getValue(), offset, TimeUnit.SECONDS); } //超时使用LastValue for (Map.Entry> measurePointsItem : measurePointsCountGroup.entrySet()) { if (finishedGroup.contains(measurePointsItem.getKey())) { continue; } notFinish.add(measurePointsItem.getKey()); HttpApiEntity httpApi = this.getHttpApi(measurePointsItem.getValue().get(0)[0].toString()); this.setOut(httpApi.getUrl(), httpApi.getCode(), result, measurePointsItem.getValue(), collectTime); } log.info("notFinish=" + JSONArray.toJSONString(notFinish)); } catch (Exception ex) { ex.printStackTrace(); } return result; } /** * 异步采集任务 */ private class Task implements Runnable { String url; String sourceName; Date collectTime; Map result; List params; CountDownLatch countDownLatch; List finishedGroup; Integer groupKey; public Task(String url, String sourceName, Map result, List params, Date collectTime, CountDownLatch countDownLatch, List finishedGroup, Integer groupKey) { this.url = url; this.sourceName = sourceName; this.result = result; this.collectTime = collectTime; this.params = params; this.countDownLatch = countDownLatch; this.finishedGroup = finishedGroup; this.groupKey = groupKey; } @Override public void run() { try { log.info("请求的Tag数量:" + params.size()); this.getByHtp(url, sourceName, result, params, collectTime); log.info("请求结束:url=" + url); } catch (Exception ex) { log.info("获取采集值失败," + ex.getMessage()); ex.printStackTrace(); } finally { countDownLatch.countDown(); finishedGroup.add(groupKey); } } /** * getTagDataByHttp * * @param url * @param sourceName * @param result * @param params * @param collectTime */ private void getByHtp(String url, String sourceName, Map result, List params, Date collectTime) { StringBuilder tagSb = new StringBuilder(); tagSb.append("["); for (int i = 0; i < params.size(); i++) { Map queryParams = new HashMap<>(3); queryParams.put(N, params.get(i)[1]); queryParams.put(D, params.get(i)[2]); queryParams.put(P, params.get(i)[3]); String jsonString = JSON.toJSONString(queryParams); tagSb.append(jsonString); if (i < params.size() - 1) { tagSb.append(","); } } tagSb.append("]"); log.info("body=====" + tagSb); String currentDate = DateUtils.format(collectTime, pattern); String responseStr = HttpUtils.sendPost(url + "/" + currentDate, tagSb.toString()); JSONObject responseObj = JSON.parseObject(responseStr); log.info("responseObj=====" + responseObj.toJSONString()); if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { JSONArray tagValueList = responseObj.getJSONArray(RESP_DATA); if (!CollectionUtils.isEmpty(tagValueList)) { for (int i = 0; i < tagValueList.size(); i++) { JSONObject item = tagValueList.getJSONObject(i); if (item.get(Q).toString().equals("G")) { result.put(TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, item.get(N).toString()), item.get(V)); } } } } } } private void getByHtp(String url, String sourceName, Map result, List params, Date collectTime) { StringBuilder tagSb = new StringBuilder(); tagSb.append("["); for (int i = 0; i < params.size(); i++) { Map queryParams = new HashMap<>(); queryParams.put(N, params.get(i)[1]); queryParams.put(D, params.get(i)[2]); queryParams.put(P, params.get(i)[3]); String jsonString = JSON.toJSONString(queryParams); tagSb.append(jsonString); if (i < params.size() - 1) { tagSb.append(","); } } tagSb.append("]"); log.info("body=====" + tagSb); String currentDate = DateUtils.format(collectTime, pattern); String responseStr = HttpUtils.sendPost(url + "/" + currentDate, tagSb.toString()); JSONObject responseObj = JSON.parseObject(responseStr); log.info("responseObj=====" + responseObj.toJSONString()); if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { JSONArray tagValueList = responseObj.getJSONArray(RESP_DATA); if (!CollectionUtils.isEmpty(tagValueList)) { for (int i = 0; i < tagValueList.size(); i++) { JSONObject item = tagValueList.getJSONObject(i); result.put(TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, item.get(N).toString()), item.get(V)); } } } } private void setOut(String url, String sourceName, Map result, List params, Date collectTime) { for (int i = 0; i < params.size(); i++) { String tagId = TIV + TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, params.get(i)[1].toString()); if (redisTemplate.hasKey(tagId)) { result.put(TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, params.get(i)[1].toString()), redisTemplate.opsForValue().get(tagId)); } } } }