Skip to content

Commit b2f0f69

Browse files
zengqiaoZQKC
zengqiao
authored and
ZQKC
committed
[Optimize]Overview页面的TopN查询ES流程优化(#823)
1、复用线程池，同时支持线程池的线程数可配置； 2、优化查询TopN指标时，可能会出现重复查询的问题； 3、处理代码扫描(SonarLint)反馈的问题；
1 parent c4fb18a commit b2f0f69

File tree

11 files changed

+400
-422
lines changed

11 files changed

+400
-422
lines changed

‎km-common/src/main/java/com/xiaojukeji/know/streaming/km/common/bean/vo/metrics/point/MetricPointVO.java‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
import io.swagger.annotations.ApiModel;
44
import io.swagger.annotations.ApiModelProperty;
5-
import lombok.AllArgsConstructor;
65
import lombok.Data;
76
import lombok.NoArgsConstructor;
87

@@ -11,7 +10,6 @@
1110
*/
1211
@Data
1312
@NoArgsConstructor
14-
@AllArgsConstructor
1513
@ApiModel(description = "指标点")
1614
public class MetricPointVO implements Comparable<MetricPointVO> {
1715
@ApiModelProperty(value = "指标名", example = "HealthScore")
@@ -26,6 +24,13 @@ public class MetricPointVO implements Comparable<MetricPointVO> {
2624
@ApiModelProperty(value = "指标值聚合方式：avg、max、min、sum")
2725
private String aggType;
2826

27+
public MetricPointVO(String name, Long timeStamp, String value, String aggType) {
28+
this.name = name;
29+
this.timeStamp = timeStamp;
30+
this.value = value;
31+
this.aggType = aggType;
32+
}
33+
2934
@Override
3035
public int compareTo(MetricPointVO o) {
3136
if(null == o){return 0;}

‎km-core/src/main/java/com/xiaojukeji/know/streaming/km/core/service/version/BaseMetricService.java‎

Lines changed: 20 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -38,40 +38,41 @@ protected void initMetricFieldAndNameList(){
3838

3939
protected abstract void initRegisterVCHandler();
4040

41-
protected <T> List<MetricMultiLinesVO> metricMap2VO(Long clusterId,
42-
Map<String/*metric*/, Map<T, List<MetricPointVO>>> map){
43-
List<MetricMultiLinesVO> multiLinesVOS = new ArrayList<>();
44-
if (map == null || map.isEmpty()) {
41+
protected <T> List<MetricMultiLinesVO> metricMap2VO(Long clusterId, Map<String/*metric*/, Map<T, List<MetricPointVO>>> metricsMap ){
42+
List<MetricMultiLinesVO> lineVOList = new ArrayList<>();
43+
if (metricsMap == null || metricsMap.isEmpty()) {
4544
// 如果为空，则直接返回
46-
return multiLinesVOS;
45+
return lineVOList;
4746
}
4847

49-
for(String metric : map.keySet()){
48+
for(Map.Entry<String/*metric*/, Map<T, List<MetricPointVO>>> entry : metricsMap.entrySet()){
5049
try {
5150
MetricMultiLinesVO multiLinesVO = new MetricMultiLinesVO();
52-
multiLinesVO.setMetricName(metric);
51+
multiLinesVO.setMetricName(entry.getKey());
5352

54-
List<MetricLineVO> metricLines = new ArrayList<>();
53+
if(null == entry.getValue() || entry.getValue().isEmpty()){
54+
continue;
55+
}
5556

56-
Map<T, List<MetricPointVO>> metricPointMap = map.get(metric);
57-
if(null == metricPointMap || metricPointMap.isEmpty()){continue;}
58-
for(Map.Entry<T, List<MetricPointVO>> entry : metricPointMap.entrySet()){
57+
List<MetricLineVO> metricLines = new ArrayList<>();
58+
entry.getValue().entrySet().forEach(resNameAndMetricsEntry -> {
5959
MetricLineVO metricLineVO = new MetricLineVO();
60-
metricLineVO.setName(entry.getKey().toString());
61-
metricLineVO.setMetricName(metric);
62-
metricLineVO.setMetricPoints(entry.getValue());
60+
metricLineVO.setName(resNameAndMetricsEntry.getKey().toString());
61+
metricLineVO.setMetricName(entry.getKey());
62+
metricLineVO.setMetricPoints(resNameAndMetricsEntry.getValue());
6363

6464
metricLines.add(metricLineVO);
65-
}
65+
});
6666

6767
multiLinesVO.setMetricLines(metricLines);
68-
multiLinesVOS.add(multiLinesVO);
69-
}catch (Exception e){
70-
LOGGER.error("method=metricMap2VO||cluster={}||msg=exception!", clusterId, e);
68+
69+
lineVOList.add(multiLinesVO);
70+
} catch (Exception e){
71+
LOGGER.error("method=metricMap2VO||clusterId={}||msg=exception!", clusterId, e);
7172
}
7273
}
7374

74-
return multiLinesVOS;
75+
return lineVOList;
7576
}
7677

7778
/**

‎km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/ESTPService.java‎

Lines changed: 41 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,41 @@
1+
package com.xiaojukeji.know.streaming.km.persistence.es;
2+
3+
import com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil;
4+
import lombok.NoArgsConstructor;
5+
import org.springframework.beans.factory.annotation.Value;
6+
import org.springframework.stereotype.Service;
7+
8+
import javax.annotation.PostConstruct;
9+
10+
/**
11+
* 处理ES请求的线程池
12+
*/
13+
@Service
14+
@NoArgsConstructor
15+
public class ESTPService {
16+
@Value("${thread-pool.es.search.thread-num:10}")
17+
private Integer esSearchThreadCnt;
18+
19+
@Value("${thread-pool.es.search.queue-size:5000}")
20+
private Integer esSearchThreadQueueSize;
21+
22+
private FutureWaitUtil<Object> searchESTP;
23+
24+
@PostConstruct
25+
private void init() {
26+
searchESTP = FutureWaitUtil.init(
27+
"SearchESTP",
28+
esSearchThreadCnt,
29+
esSearchThreadCnt,
30+
esSearchThreadQueueSize
31+
);
32+
}
33+
34+
public void submitSearchTask(String taskName, Integer timeoutUnisMs, Runnable runnable) {
35+
searchESTP.runnableTask(taskName, timeoutUnisMs, runnable);
36+
}
37+
38+
public void waitExecute() {
39+
searchESTP.waitExecute();
40+
}
41+
}

‎km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/dao/BaseMetricESDAO.java‎

Lines changed: 12 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -12,7 +12,9 @@
1212
import com.xiaojukeji.know.streaming.km.common.utils.CommonUtils;
1313
import com.xiaojukeji.know.streaming.km.common.utils.EnvUtil;
1414
import com.xiaojukeji.know.streaming.km.common.utils.IndexNameUtils;
15+
import com.xiaojukeji.know.streaming.km.common.utils.ValidateUtils;
1516
import com.xiaojukeji.know.streaming.km.persistence.es.BaseESDAO;
17+
import com.xiaojukeji.know.streaming.km.persistence.es.ESTPService;
1618
import com.xiaojukeji.know.streaming.km.persistence.es.dsls.DslConstant;
1719
import com.xiaojukeji.know.streaming.km.persistence.es.template.TemplateLoaderUtil;
1820
import lombok.NoArgsConstructor;
@@ -48,6 +50,9 @@ public class BaseMetricESDAO extends BaseESDAO {
4850
@Autowired
4951
private TemplateLoaderUtil templateLoaderUtil;
5052

53+
@Autowired
54+
protected ESTPService esTPService;
55+
5156
/**
5257
* es 地址
5358
*/
@@ -364,21 +369,16 @@ public void buildAggsDslMap(String aggType, StringBuilder sb, String metricName,
364369
sb.append(str, 1, str.length() - 1);
365370
}
366371

367-
protected Map<String, ESAggr> checkBucketsAndHitsOfResponseAggs(ESQueryResponse response){
368-
if(null == response || null == response.getAggs()){
369-
return null;
370-
}
371-
372-
Map<String, ESAggr> esAggrMap = response.getAggs().getEsAggrMap();
373-
if (null == esAggrMap || null == esAggrMap.get(HIST)) {
374-
return null;
375-
}
376-
377-
if(CollectionUtils.isEmpty(esAggrMap.get(HIST).getBucketList())){
372+
protected Map<String, ESAggr> checkBucketsAndHitsOfResponseAggs(ESQueryResponse response) {
373+
if(null == response
374+
|| null == response.getAggs()
375+
|| null == response.getAggs().getEsAggrMap()
376+
|| null == response.getAggs().getEsAggrMap().get(HIST)
377+
|| ValidateUtils.isEmptyList(response.getAggs().getEsAggrMap().get(HIST).getBucketList())) {
378378
return null;
379379
}
380380

381-
return esAggrMap;
381+
return response.getAggs().getEsAggrMap();
382382
}
383383

384384
protected int handleESQueryResponseCount(ESQueryResponse response){

‎km-persistence/src/main/java/com/xiaojukeji/know/streaming/km/persistence/es/dao/BrokerMetricESDAO.java‎

Lines changed: 47 additions & 71 deletions
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,10 @@
66
import com.google.common.collect.Table;
77
import com.xiaojukeji.know.streaming.km.common.bean.po.metrice.BrokerMetricPO;
88
import com.xiaojukeji.know.streaming.km.common.bean.vo.metrics.point.MetricPointVO;
9-
import com.xiaojukeji.know.streaming.km.common.utils.FutureWaitUtil;
109
import com.xiaojukeji.know.streaming.km.common.utils.MetricsUtils;
1110
import com.xiaojukeji.know.streaming.km.common.utils.Tuple;
1211
import com.xiaojukeji.know.streaming.km.persistence.es.dsls.DslConstant;
1312
import org.springframework.stereotype.Component;
14-
import org.springframework.util.CollectionUtils;
1513

1614
import javax.annotation.PostConstruct;
1715
import java.util.*;
@@ -29,8 +27,6 @@ public void init() {
2927
register( this);
3028
}
3129

32-
protected FutureWaitUtil<Void> queryFuture = FutureWaitUtil.init("BrokerMetricESDAO", 4,8, 500);
33-
3430
/**
3531
* 获取集群 clusterId 中 brokerId 最新的统计指标
3632
*/
@@ -140,7 +136,7 @@ public BrokerMetricPO getBrokerLatestMetrics(Long clusterId, Integer brokerId){
140136
aggDsl
141137
);
142138

143-
queryFuture.runnableTask(
139+
esTPService.submitSearchTask(
144140
String.format("class=BrokerMetricESDAO||method=listBrokerMetricsByBrokerIds||ClusterPhyId=%d", clusterPhyId),
145141
5000,
146142
() -> {
@@ -163,7 +159,7 @@ public BrokerMetricPO getBrokerLatestMetrics(Long clusterId, Integer brokerId){
163159
}
164160
}
165161

166-
queryFuture.waitExecute();
162+
esTPService.waitExecute();
167163

168164
return table;
169165
}
@@ -220,106 +216,86 @@ public Map<String, List<Long>> getTopNBrokerIds(Long clusterPhyId, List<String>
220216
return metricMap;
221217
}
222218

223-
private Map<String, List<MetricPointVO>> handleListESQueryResponse(ESQueryResponse response, List<String> metrics, String aggType){
219+
private Map<String, List<MetricPointVO>> handleListESQueryResponse(ESQueryResponse response, List<String> metricNameList, String aggType){
224220
Map<String, List<MetricPointVO>> metricMap = new HashMap<>();
225221

226-
if(null == response || null == response.getAggs()){
227-
return metricMap;
228-
}
229-
230-
Map<String, ESAggr> esAggrMap = response.getAggs().getEsAggrMap();
231-
if (null == esAggrMap || null == esAggrMap.get(HIST)) {
232-
return metricMap;
233-
}
234-
235-
if(CollectionUtils.isEmpty(esAggrMap.get(HIST).getBucketList())){
222+
Map<String, ESAggr> esAggrMap = this.checkBucketsAndHitsOfResponseAggs(response);
223+
if (esAggrMap == null) {
236224
return metricMap;
237225
}
238226

239-
for(String metric : metrics){
227+
for(String metricName : metricNameList){
240228
List<MetricPointVO> metricPoints = new ArrayList<>();
241229

242-
esAggrMap.get(HIST).getBucketList().forEach( esBucket -> {
230+
esAggrMap.get(HIST).getBucketList().forEach(esBucket -> {
243231
try {
244-
if (null != esBucket.getUnusedMap().get(KEY)) {
245-
Long timestamp = Long.valueOf(esBucket.getUnusedMap().get(KEY).toString());
246-
Object value = esBucket.getAggrMap().get(metric).getUnusedMap().get(VALUE);
247-
if(null == value){return;}
248-
249-
MetricPointVO metricPoint = new MetricPointVO();
250-
metricPoint.setAggType(aggType);
251-
metricPoint.setTimeStamp(timestamp);
252-
metricPoint.setValue(value.toString());
253-
metricPoint.setName(metric);
254-
255-
metricPoints.add(metricPoint);
256-
}else {
257-
LOGGER.info("");
232+
if (null == esBucket.getUnusedMap().get(KEY)) {
233+
return;
258234
}
259-
}catch (Exception e){
260-
LOGGER.error("metric={}||errMsg=exception!", metric, e);
235+
236+
Long timestamp = Long.valueOf(esBucket.getUnusedMap().get(KEY).toString());
237+
Object value = esBucket.getAggrMap().get(metricName).getUnusedMap().get(VALUE);
238+
if(null == value) {
239+
return;
240+
}
241+
242+
metricPoints.add(new MetricPointVO(metricName, timestamp, value.toString(), aggType));
243+
} catch (Exception e){
244+
LOGGER.error("method=handleListESQueryResponse||metricName={}||errMsg=exception!", metricName, e);
261245
}
262246
} );
263247

264-
metricMap.put(metric, optimizeMetricPoints(metricPoints));
248+
metricMap.put(metricName, optimizeMetricPoints(metricPoints));
265249
}
266250

267251
return metricMap;
268252
}
269253

270-
private Map<String, List<Long>> handleTopBrokerESQueryResponse(ESQueryResponse response, List<String> metrics, int topN){
254+
private Map<String, List<Long>> handleTopBrokerESQueryResponse(ESQueryResponse response, List<String> metricNameList, int topN) {
271255
Map<String, List<Long>> ret = new HashMap<>();
272256

273-
if(null == response || null == response.getAggs()){
274-
return ret;
275-
}
276-
277-
Map<String, ESAggr> esAggrMap = response.getAggs().getEsAggrMap();
278-
if (null == esAggrMap || null == esAggrMap.get(HIST)) {
279-
return ret;
280-
}
281-
282-
if(CollectionUtils.isEmpty(esAggrMap.get(HIST).getBucketList())){
257+
Map<String, ESAggr> esAggrMap = this.checkBucketsAndHitsOfResponseAggs(response);
258+
if (esAggrMap == null) {
283259
return ret;
284260
}
285261

286-
Map<String, List<Tuple<Long, Double>>> metricBrokerValueMap = new HashMap<>();
262+
Map<String, List<Tuple<Long, Double>>> metricNameBrokerValueMap = new HashMap<>();
287263

288264
//1、先获取每个指标对应的所有brokerIds以及指标的值
289-
for(String metric : metrics) {
290-
esAggrMap.get(HIST).getBucketList().forEach( esBucket -> {
265+
for(String metricName : metricNameList) {
266+
esAggrMap.get(HIST).getBucketList().forEach(esBucket -> {
291267
try {
292-
if (null != esBucket.getUnusedMap().get(KEY)) {
293-
Long brokerId = Long.valueOf(esBucket.getUnusedMap().get(KEY).toString());
294-
Object value = esBucket.getAggrMap().get(HIST).getBucketList().get(0).getAggrMap()
295-
.get(metric).getUnusedMap().get(VALUE);
296-
if(null == value){return;}
297-
298-
List<Tuple<Long, Double>> brokerValue = (null == metricBrokerValueMap.get(metric)) ?
299-
new ArrayList<>() : metricBrokerValueMap.get(metric);
268+
if (null == esBucket.getUnusedMap().get(KEY)) {
269+
return;
270+
}
300271

301-
brokerValue.add(new Tuple<>(brokerId, Double.valueOf(value.toString())));
302-
metricBrokerValueMap.put(metric, brokerValue);
272+
Long brokerId = Long.valueOf(esBucket.getUnusedMap().get(KEY).toString());
273+
Object value = esBucket.getAggrMap().get(HIST).getBucketList().get(0).getAggrMap().get(metricName).getUnusedMap().get(VALUE);
274+
if(null == value) {
275+
return;
303276
}
304-
}catch (Exception e){
305-
LOGGER.error("metrice={}||errMsg=exception!", metric, e);
277+
278+
metricNameBrokerValueMap.putIfAbsent(metricName, new ArrayList<>());
279+
metricNameBrokerValueMap.get(metricName).add(new Tuple<>(brokerId, Double.valueOf(value.toString())));
280+
} catch (Exception e) {
281+
LOGGER.error("method=handleTopBrokerESQueryResponse||metric={}||errMsg=exception!", metricName, e);
306282
}
307-
} );
283+
});
308284
}
309285

310286
//2、对每个指标的broker按照指标值排序，并截取前topN个brokerIds
311-
for(String metric : metricBrokerValueMap.keySet()){
312-
List<Tuple<Long, Double>> brokerValue = metricBrokerValueMap.get(metric);
287+
for(Map.Entry<String, List<Tuple<Long, Double>>> entry : metricNameBrokerValueMap.entrySet()){
288+
entry.getValue().sort((o1, o2) -> {
289+
if(null == o1 || null == o2){
290+
return 0;
291+
}
313292

314-
brokerValue.sort((o1, o2) -> {
315-
if(null == o1 || null == o2){return 0;}
316293
return o2.getV2().compareTo(o1.getV2());
317294
} );
318295

319-
List<Tuple<Long, Double>> temp = (brokerValue.size() > topN) ? brokerValue.subList(0, topN) : brokerValue;
320-
List<Long> brokerIds = temp.stream().map(t -> t.getV1()).collect( Collectors.toList());
321-
322-
ret.put(metric, brokerIds);
296+
// 获取TopN的Broker
297+
List<Long> brokerIdList = entry.getValue().subList(0, Math.min(topN, entry.getValue().size())).stream().map(elem -> elem.getV1()).collect(Collectors.toList());
298+
ret.put(entry.getKey(), brokerIdList);
323299
}
324300

325301
return ret;

0 commit comments

Comments
 (0)