iailab-plat.git

潘志宝

7 天以前 bbe7acfbe5a4c08d6edc91eaf81dcecf9d630e18

提交 \| 用户 \| 时间
52487d	1	package com.iailab.module.data.channel.http.collector.ihdb;
L	2
	3	import com.alibaba.fastjson.JSON;
	4	import com.alibaba.fastjson.JSONArray;
	5	import com.alibaba.fastjson.JSONObject;
e8ad66	6	import com.google.gson.Gson;
52487d	7	import com.iailab.framework.common.constant.CommonConstant;
L	8	import com.iailab.module.data.channel.http.entity.HttpApiEntity;
	9	import com.iailab.module.data.channel.http.service.HttpApiService;
	10	import com.iailab.module.data.common.enums.DataSourceType;
	11	import com.iailab.module.data.common.utils.DateUtils;
d41f14	12	import com.iailab.module.data.common.utils.HttpRequest;
52487d	13	import com.iailab.module.data.common.utils.TagUtils;
L	14	import lombok.extern.slf4j.Slf4j;
2228b6	15	import org.apache.commons.lang3.StringUtils;
52487d	16	import org.springframework.beans.factory.annotation.Autowired;
2e7f34	17	import org.springframework.data.redis.core.BoundHashOperations;
D	18	import org.springframework.data.redis.core.RedisTemplate;
52487d	19	import org.springframework.stereotype.Component;
L	20	import org.springframework.util.CollectionUtils;
	21
	22	import java.math.BigDecimal;
e8ad66	23	import java.math.RoundingMode;
0a2804	24	import java.sql.Timestamp;
潘	25	import java.text.ParseException;
	26	import java.text.SimpleDateFormat;
	27	import java.time.LocalDateTime;
52487d	28	import java.util.*;
0a2804	29	import java.util.concurrent.*;
潘	30	import java.util.function.Function;
	31	import java.util.stream.Collectors;
52487d	32
L	33	/**
	34	* iHyperDB采集
d41f14	35	*
52487d	36	* @author lirm
L	37	* @Description
	38	* @createTime 2024年10月16日
	39	*/
	40	@Slf4j
	41	@Component
d41f14	42	public class HttpCollectorForIhd {
0a2804	43	private static Map<String, HttpApiEntity> apiMap = new HashMap<>();
52487d	44
L	45	@Autowired
	46	private HttpApiService httpApiService;
2e7f34	47
D	48	@Autowired
	49	private RedisTemplate redisTemplate;
e8ad66	50
0a2804	51	ThreadPoolExecutor threadPool = new ThreadPoolExecutor(18, 36, 30, TimeUnit.SECONDS,
潘	52	new ArrayBlockingQueue<Runnable>(36), new ThreadPoolExecutor.AbortPolicy());
	53
d41f14	54	private static final String STA_TRUE = "true";
52487d	55
1e563e	56	private static final int GROUP_MAX_COUNT = 300;
0a2804	57
潘	58	private static final int MAX_WAIT = 30;
2f03e2	59
潘	60	private static final String pattern = "yyyyMMddHHmm00";
	61
	62	private static final String IS_SUCCESS = "isSuccess";
	63
	64	/**
	65	* tagName
	66	*/
	67	private static final String N = "n";
	68
	69	/**
	70	* dimension
	71	*/
	72	private static final String D = "d";
	73
	74	/**
	75	* 类型
	76	*/
	77	private static final String P = "p";
	78
	79	/**
	80	* dataValue
	81	*/
	82	private static final String V = "v";
	83
	84	/**
	85	* dataTime
	86	*/
	87	private static final String T = "t";
52487d	88
L	89	private HttpApiEntity getHttpApi(String id) {
	90	if (apiMap.containsKey(id)) {
	91	return apiMap.get(id);
	92	}
	93	HttpApiEntity httpApi = httpApiService.info(id);
	94	apiMap.put(id, httpApi);
	95	return httpApi;
	96	}
	97
	98	public BigDecimal getTagValue(String sourceId, String tagNo, Integer dimension, String valueType) {
	99	BigDecimal value = CommonConstant.BAD_VALUE;
	100	HttpApiEntity httpApi = this.getHttpApi(sourceId);
	101	StringBuilder tagSb = new StringBuilder();
	102	tagSb.append("[");
	103	Map<String, Object> queryParams = new HashMap<>();
2f03e2	104	queryParams.put(P, valueType);
潘	105	queryParams.put(D, dimension);
	106	queryParams.put(N, tagNo);
d41f14	107	String jsonString = JSON.toJSONString(queryParams);
52487d	108	tagSb.append(jsonString);
L	109	tagSb.append("]");
2f03e2	110	String currentDate = DateUtils.format(new Date(), pattern);
d41f14	111	String responseStr = HttpRequest.sendPost(httpApi.getUrl() + "/" + currentDate, tagSb.toString());
52487d	112	JSONObject responseObj = JSON.parseObject(responseStr);
2f03e2	113	if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) {
52487d	114	JSONArray tagValueList = responseObj.getJSONArray("data");
L	115	if (!CollectionUtils.isEmpty(tagValueList)) {
	116	for (int i = 0; i < tagValueList.size(); i++) {
	117	JSONObject item = tagValueList.getJSONObject(i);
2f03e2	118	value = new BigDecimal(item.get(V).toString());
52487d	119	}
L	120	}
	121	}
	122	return value;
	123	}
	124
2f03e2	125	public Map<String, Object> getLastValues(String sourceId, List<String> tagNames) {
e8ad66	126	Map<String, Object> result = new HashMap<>();
2f03e2	127	HttpApiEntity httpApi = this.getHttpApi(sourceId);
2e7f34	128	try {
D	129	if (CollectionUtils.isEmpty(tagNames)) {
	130	return result;
	131	}
	132	List<String> noCacheTagNames = new ArrayList<>();//未缓存的tag
	133	for (int i = 0; i < tagNames.size(); i++) {
	134	//先查缓存
	135	BoundHashOperations<String, String, Object> ops = redisTemplate.boundHashOps(tagNames.get(i));
2f03e2	136	if (ops.get(V) != null) {
潘	137	BigDecimal value = new BigDecimal(ops.get(V).toString());
2e7f34	138	result.put(tagNames.get(i), value.setScale(3, RoundingMode.HALF_UP));
D	139	} else {
	140	noCacheTagNames.add(tagNames.get(i));
	141	}
	142	}
	143	if (CollectionUtils.isEmpty(noCacheTagNames)) {
	144	log.info("全部读取缓存");
	145	return result;
	146	}
	147	log.info("查询未缓存的数据");
	148	Gson gson = new Gson();
	149	String tagSb = gson.toJson(noCacheTagNames);
	150	log.info("body=====" + tagSb);
2f03e2	151	String currentDate = DateUtils.format(new Date(), pattern);
2e7f34	152	String responseStr = "";
2f03e2	153	responseStr = HttpRequest.sendPost(httpApi.getUrl().replace("getPointdatasAvg", "getPointslast") + "/" + currentDate, tagSb);
2e7f34	154	JSONObject responseObj = JSON.parseObject(responseStr);
2f03e2	155	if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) {
2e7f34	156	JSONArray tagValueList = responseObj.getJSONArray("data");
D	157	if (!CollectionUtils.isEmpty(tagValueList)) {
	158	for (int i = 0; i < tagValueList.size(); i++) {
	159	JSONObject item = tagValueList.getJSONObject(i);
2f03e2	160	if (item.get(V) != null) {
2e7f34	161	//存缓存
2f03e2	162	BoundHashOperations<String, String, Object> ops = redisTemplate.boundHashOps(item.get(V).toString());
潘	163	ops.put(V, item.get(V).toString());
2e7f34	164	//设置过期时间
2f03e2	165	redisTemplate.expire(item.get(N).toString(), 10, TimeUnit.SECONDS);
2e7f34	166	//把查询到的数据插入结果集
2f03e2	167	BigDecimal value = new BigDecimal(item.get(V).toString());
潘	168	result.put(item.get(N).toString(), value.setScale(3, RoundingMode.HALF_UP));
2e7f34	169	} else {
2f03e2	170	result.put(item.get(N).toString(), CommonConstant.BAD_VALUE);
2e7f34	171	}
e8ad66	172	}
D	173	}
	174	}
2e7f34	175
D	176	} catch (Exception ex) {
	177	log.info("getCurrentValue异常");
	178	ex.printStackTrace();
	179	throw ex;
e8ad66	180	}
D	181	return result;
	182	}
	183
0a2804	184	public Map<String, Object> getTagValues(List<Object[]> params, Date collectTime) {
潘	185	Map<String, Object> result = new HashMap<>();
52487d	186	if (CollectionUtils.isEmpty(params)) {
L	187	return new HashMap<>();
	188	}
0a2804	189	try {
潘	190	Map<Integer, List<Object[]>> measurePointsCountGroup = new HashMap<>();
	191	int pointListSize = params.size();
	192	int groupCount = pointListSize / GROUP_MAX_COUNT + ((pointListSize % GROUP_MAX_COUNT) > 0 ? 1 : 0);
	193	log.info("groupCount=" + groupCount);
	194	for (int i = 0; i < groupCount; i++) {
	195	int end = (i + 1) * GROUP_MAX_COUNT;
	196	if (end > pointListSize) {
	197	end = pointListSize;
	198	}
	199	measurePointsCountGroup.put(i, params.subList(i * GROUP_MAX_COUNT, end));
	200	}
	201	log.info("measurePointsCountGroup.size()=" + measurePointsCountGroup.size());
	202	result = new ConcurrentHashMap<>(params.size());
	203	CountDownLatch countDownLatch = new CountDownLatch(measurePointsCountGroup.size());
	204	for (Map.Entry<Integer, List<Object[]>> measurePointsItem : measurePointsCountGroup.entrySet()) {
	205	HttpApiEntity httpApi = this.getHttpApi(measurePointsItem.getValue().get(0)[0].toString());
	206	// 并发
1e563e	207	Thread.sleep(1000);
0a2804	208	threadPool.submit(new Task(httpApi.getUrl(), httpApi.getCode(), result, measurePointsItem.getValue(),
潘	209	collectTime, countDownLatch));
	210	// 顺序
	211	//this.getByHtp(result, measurePointsItem.getValue(), collectTime);
52487d	212	}
0a2804	213	countDownLatch.await(MAX_WAIT, TimeUnit.SECONDS);
潘	214
	215	} catch (Exception ex) {
	216	ex.printStackTrace();
52487d	217	}
L	218	return result;
	219	}
	220
0a2804	221	/**
潘	222	* 异步采集任务
	223	*/
	224	private class Task implements Runnable {
	225	String url;
	226	String sourceName;
	227	Date collectTime;
	228	Map<String, Object> result;
	229	List<Object[]> params;
	230	CountDownLatch countDownLatch;
	231
	232	public Task(String url, String sourceName, Map<String, Object> result, List<Object[]> params,
	233	Date collectTime, CountDownLatch countDownLatch) {
	234	this.url = url;
	235	this.sourceName = sourceName;
	236	this.result = result;
	237	this.collectTime = collectTime;
	238	this.params = params;
	239	this.countDownLatch = countDownLatch;
	240	}
	241
	242	@Override
	243	public void run() {
	244	try {
	245	log.info("请求的Tag数量：" + params.size());
	246	this.getByHtp(url, sourceName, result, params, collectTime);
	247	log.info("请求结束：url=" + url);
	248	} catch (Exception ex) {
	249	log.info("获取采集值失败," + ex.getMessage());
	250	ex.printStackTrace();
	251	} finally {
	252	countDownLatch.countDown();
	253	}
	254	}
	255
	256	/**
	257	* getTagDataByHttp
	258	*
	259	* @param url
	260	* @param sourceName
	261	* @param result
	262	* @param params
	263	* @param collectTime
	264	*/
	265	private void getByHtp(String url, String sourceName, Map<String, Object> result, List<Object[]> params, Date collectTime) {
	266	StringBuilder tagSb = new StringBuilder();
	267	tagSb.append("[");
	268	for (int i = 0; i < params.size(); i++) {
2228b6	269	Map<String, Object> queryParams = new HashMap<>(3);
0a2804	270	queryParams.put(N, params.get(i)[1]);
潘	271	queryParams.put(D, params.get(i)[2]);
	272	queryParams.put(P, params.get(i)[3]);
	273	String jsonString = JSON.toJSONString(queryParams);
	274	tagSb.append(jsonString);
	275	if (i < params.size() - 1) {
	276	tagSb.append(",");
	277	}
	278	}
	279	tagSb.append("]");
	280	log.info("body=====" + tagSb);
	281	String currentDate = DateUtils.format(collectTime, pattern);
	282	String responseStr = HttpRequest.sendPost(url + "/" + currentDate, tagSb.toString());
	283	JSONObject responseObj = JSON.parseObject(responseStr);
	284	log.info("responseObj=====" + responseObj.toJSONString());
	285	if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) {
	286	JSONArray tagValueList = responseObj.getJSONArray("data");
	287	if (!CollectionUtils.isEmpty(tagValueList)) {
	288	for (int i = 0; i < tagValueList.size(); i++) {
	289	JSONObject item = tagValueList.getJSONObject(i);
	290	result.put(TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, item.get(N).toString()), item.get(V));
	291	}
	292	}
	293	}
	294	}
	295	}
	296
	297	private void getByHtp(String url, String sourceName, Map<String, Object> result, List<Object[]> params, Date collectTime) {
52487d	298	StringBuilder tagSb = new StringBuilder();
L	299	tagSb.append("[");
d41f14	300	for (int i = 0; i < params.size(); i++) {
52487d	301	Map<String, Object> queryParams = new HashMap<>();
2f03e2	302	queryParams.put(N, params.get(i)[1]);
潘	303	queryParams.put(D, params.get(i)[2]);
	304	queryParams.put(P, params.get(i)[3]);
d41f14	305	String jsonString = JSON.toJSONString(queryParams);
52487d	306	tagSb.append(jsonString);
L	307	if (i < params.size() - 1) {
	308	tagSb.append(",");
	309	}
	310	}
	311	tagSb.append("]");
0a2804	312	log.info("body=====" + tagSb);
潘	313	String currentDate = DateUtils.format(collectTime, pattern);
	314	String responseStr = HttpRequest.sendPost(url + "/" + currentDate, tagSb.toString());
52487d	315	JSONObject responseObj = JSON.parseObject(responseStr);
L	316	log.info("responseObj=====" + responseObj.toJSONString());
2f03e2	317	if (STA_TRUE.equals(responseObj.get(IS_SUCCESS).toString())) {
52487d	318	JSONArray tagValueList = responseObj.getJSONArray("data");
L	319	if (!CollectionUtils.isEmpty(tagValueList)) {
	320	for (int i = 0; i < tagValueList.size(); i++) {
	321	JSONObject item = tagValueList.getJSONObject(i);
0a2804	322	result.put(TagUtils.genTagId(DataSourceType.HTTP.getCode(), sourceName, item.get(N).toString()), item.get(V));
52487d	323	}
L	324	}
	325	}
	326	}
	327	}