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.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 = 30; private static final String pattern = "yyyyMMddHHmm00"; private static final String IS_SUCCESS = "isSuccess"; /** * 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"; 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 = HttpRequest.sendPost(httpApi.getUrl() + "/" + currentDate, tagSb.toString()); JSONObject responseObj = JSON.parseObject(responseStr); if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { JSONArray tagValueList = responseObj.getJSONArray("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 (int i = 0; i < tagNames.size(); i++) { //先查缓存 BoundHashOperations ops = redisTemplate.boundHashOps(tagNames.get(i)); if (ops.get(V) != null) { BigDecimal value = new BigDecimal(ops.get(V).toString()); result.put(tagNames.get(i), value.setScale(3, RoundingMode.HALF_UP)); } else { noCacheTagNames.add(tagNames.get(i)); } } if (CollectionUtils.isEmpty(noCacheTagNames)) { log.info("全部读取缓存"); return result; } log.info("查询未缓存的数据"); Gson gson = new Gson(); String tagSb = gson.toJson(noCacheTagNames); log.info("body=====" + tagSb); String currentDate = DateUtils.format(new Date(), pattern); String responseStr = ""; responseStr = HttpRequest.sendPost(httpApi.getUrl().replace("getPointdatasAvg", "getPointslast") + "/" + currentDate, tagSb); JSONObject responseObj = JSON.parseObject(responseStr); if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) { JSONArray tagValueList = responseObj.getJSONArray("data"); if (!CollectionUtils.isEmpty(tagValueList)) { for (int i = 0; i < tagValueList.size(); i++) { JSONObject item = tagValueList.getJSONObject(i); if (item.get(V) != null) { //存缓存 BoundHashOperations ops = redisTemplate.boundHashOps(item.get(V).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)); } else { result.put(item.get(N).toString(), CommonConstant.BAD_VALUE); } } } } } 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()); for (Map.Entry> measurePointsItem : measurePointsCountGroup.entrySet()) { HttpApiEntity httpApi = this.getHttpApi(measurePointsItem.getValue().get(0)[0].toString()); // 并发 Thread.sleep(1000); threadPool.submit(new Task(httpApi.getUrl(), httpApi.getCode(), result, measurePointsItem.getValue(), collectTime, countDownLatch)); // 顺序 //this.getByHtp(result, measurePointsItem.getValue(), collectTime); } countDownLatch.await(MAX_WAIT, TimeUnit.SECONDS); } catch (Exception ex) { ex.printStackTrace(); } return result; } /** * 异步采集任务 */ private class Task implements Runnable { String url; String sourceName; Date collectTime; Map result; List params; CountDownLatch countDownLatch; public Task(String url, String sourceName, Map result, List params, Date collectTime, CountDownLatch countDownLatch) { this.url = url; this.sourceName = sourceName; this.result = result; this.collectTime = collectTime; this.params = params; this.countDownLatch = countDownLatch; } @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(); } } /** * 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++) { if (StringUtils.isBlank(params.get(i)[1].toString()) || StringUtils.isBlank(params.get(i)[2].toString()) || StringUtils.isBlank(params.get(i)[3].toString())) { continue; } 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 = HttpRequest.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("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 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 = HttpRequest.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("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)); } } } } }