簡(jiǎn)歷寫(xiě)著熟悉 Dubbo，居然連 Dubbo 線程池監(jiān)控都不知道？
閱讀本文大概需要 8 分鐘。
來(lái)自：網(wǎng)絡(luò)
Dubbo底層對(duì)于線程池的查看
package org.idea.dubbo.monitor.core.collect;
import org.apache.dubbo.common.extension.ExtensionLoader;
import org.apache.dubbo.common.threadpool.manager.DefaultExecutorRepository;
import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
import java.lang.reflect.Field;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
/**
* @Author idea
* @Date created in 7:04 下午 2022/6/29
*/
public class DubboThreadPoolCollector {
/**
* 獲取Dubbo的線程池
* @return
*/
public static ThreadPoolExecutor getDubboThreadPoolInfo(){
//dubbo線程池?cái)?shù)量監(jiān)控
try {
ExtensionLoader<ExecutorRepository> executorRepositoryExtensionLoader = ExtensionLoader.getExtensionLoader(ExecutorRepository.class);
DefaultExecutorRepository defaultExecutorRepository = (DefaultExecutorRepository) executorRepositoryExtensionLoader.getDefaultExtension();
Field dataField = defaultExecutorRepository.getClass().getDeclaredField("data");
dataField.setAccessible(true);
ConcurrentMap<String, ConcurrentMap<Integer, ExecutorService>> data = (ConcurrentMap<String, ConcurrentMap<Integer, ExecutorService>>) dataField.get(defaultExecutorRepository);
ConcurrentMap<Integer, ExecutorService> executorServiceConcurrentMap = data.get("java.util.concurrent.ExecutorService");
//獲取到默認(rèn)的線程池模型
ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor) executorServiceConcurrentMap.get(9090);
return threadPoolExecutor;
} catch (Exception e) {
e.printStackTrace();
}
return null;
}
}
采集數(shù)據(jù)
后臺(tái)開(kāi)啟一個(gè)定時(shí)任務(wù)，然后每秒都查詢一下線程池的參數(shù)信息。 每次有請(qǐng)求抵達(dá)provider的時(shí)候，就查看一些線程池的參數(shù)信息。
package org.idea.dubbo.monitor.core.filter;
import org.apache.dubbo.common.constants.CommonConstants;
import org.apache.dubbo.common.extension.Activate;
import org.apache.dubbo.rpc.*;
import org.idea.dubbo.monitor.core.DubboMonitorHandler;
import java.util.concurrent.ThreadPoolExecutor;
import static org.idea.dubbo.monitor.core.config.CommonCache.DUBBO_INFO_STORE_CENTER;
/**
* @Author idea
* @Date created in 2:33 下午 2022/7/1
*/
@Activate(group = CommonConstants.PROVIDER)
public class DubboRecordFilter implements Filter {
@Override
public Result invoke(Invoker<?> invoker, Invocation invocation) throws RpcException {
ThreadPoolExecutor threadPoolExecutor = DubboMonitorHandler.getDubboThreadPoolInfo();
//請(qǐng)求的時(shí)候趣統(tǒng)計(jì)線程池，當(dāng)請(qǐng)求量太小的時(shí)候，這塊的數(shù)據(jù)可能不準(zhǔn)確，但是如果請(qǐng)求量大的話，就接近準(zhǔn)確了
DUBBO_INFO_STORE_CENTER.reportInfo(9090,threadPoolExecutor.getActiveCount(),threadPoolExecutor.getQueue().size());
return invoker.invoke(invocation);
}
}
dubboRecordFilter=org.idea.dubbo.monitor.core.filter.DubboRecordFilter
package org.idea.dubbo.monitor.core.collect;
import org.idea.dubbo.monitor.core.bo.DubboInfoStoreBO;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* Dubbo數(shù)據(jù)存儲(chǔ)中心
*
* @Author idea
* @Date created in 11:15 上午 2022/7/1
*/
public class DubboInfoStoreCenter {
private static Map<Integer, DubboInfoStoreBO> dubboInfoStoreBOMap = new ConcurrentHashMap<>();
public void reportInfo(Integer port, Integer corePoolSize, Integer queueLength) {
synchronized (this) {
DubboInfoStoreBO dubboInfoStoreBO = dubboInfoStoreBOMap.get(port);
if (dubboInfoStoreBO != null) {
boolean hasChange = false;
int currentMaxPoolSize = dubboInfoStoreBO.getMaxCorePoolSize();
int currentMaxQueueLength = dubboInfoStoreBO.getMaxCorePoolSize();
if (corePoolSize > currentMaxPoolSize) {
dubboInfoStoreBO.setMaxCorePoolSize(corePoolSize);
hasChange = true;
}
if (queueLength > currentMaxQueueLength) {
dubboInfoStoreBO.setMaxQueueLength(queueLength);
hasChange = true;
}
if (hasChange) {
dubboInfoStoreBOMap.put(port, dubboInfoStoreBO);
}
} else {
dubboInfoStoreBO = new DubboInfoStoreBO();
dubboInfoStoreBO.setMaxQueueLength(queueLength);
dubboInfoStoreBO.setMaxCorePoolSize(corePoolSize);
dubboInfoStoreBOMap.put(port, dubboInfoStoreBO);
}
}
}
public DubboInfoStoreBO getInfo(Integer port){
return dubboInfoStoreBOMap.get(port);
}
public void cleanInfo(Integer port) {
dubboInfoStoreBOMap.remove(port);
}
}
package org.idea.dubbo.monitor.core.config;
import org.idea.dubbo.monitor.core.store.DubboInfoStoreCenter;
/**
* @Author idea
* @Date created in 12:15 下午 2022/7/1
*/
public class CommonCache {
public static DubboInfoStoreCenter DUBBO_INFO_STORE_CENTER = new DubboInfoStoreCenter();
}
上報(bào)數(shù)據(jù)
package org.idea.dubbo.monitor.core.bo;
/**
* @Author idea
* @Date created in 7:17 下午 2022/6/29
*/
public class ThreadInfoBO {
private Integer activePoolSize;
private Integer queueLength;
private long saveTime;
public Integer getActivePoolSize() {
return activePoolSize;
}
public void setActivePoolSize(Integer activePoolSize) {
this.activePoolSize = activePoolSize;
}
public Integer getQueueLength() {
return queueLength;
}
public void setQueueLength(Integer queueLength) {
this.queueLength = queueLength;
}
public long getSaveTime() {
return saveTime;
}
public void setSaveTime(long saveTime) {
this.saveTime = saveTime;
}
@Override
public String toString() {
return "ThreadInfoBO{" +
", queueLength=" + queueLength +
", saveTime=" + saveTime +
'}';
}
}
package org.idea.dubbo.monitor.core.report;
import com.alibaba.fastjson.JSON;
import org.idea.dubbo.monitor.core.bo.DubboInfoStoreBO;
import org.idea.dubbo.monitor.core.bo.ThreadInfoBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import static org.idea.dubbo.monitor.core.config.CommonCache.DUBBO_INFO_STORE_CENTER;
/**
* @Author idea
* @Date created in 12:13 下午 2022/7/1
*/
public class DubboInfoReportHandler implements CommandLineRunner {
@Autowired
private IReportTemplate reportTemplate;
private static final Logger LOGGER = LoggerFactory.getLogger(DubboInfoReportHandler.class);
public static ExecutorService executorService = Executors.newFixedThreadPool(1);
public static int DUBBO_PORT = 9090;
@Override
public void run(String... args) throws Exception {
executorService.submit(new Runnable() {
@Override
public void run() {
while (true) {
try {
Thread.sleep(10000);
DubboInfoStoreBO dubboInfoStoreBO = DUBBO_INFO_STORE_CENTER.getInfo(DUBBO_PORT);
ThreadInfoBO threadInfoBO = new ThreadInfoBO();
threadInfoBO.setSaveTime(System.currentTimeMillis());
if(dubboInfoStoreBO!=null){
threadInfoBO.setQueueLength(dubboInfoStoreBO.getMaxQueueLength());
threadInfoBO.setActivePoolSize(dubboInfoStoreBO.getMaxCorePoolSize());
} else {
//這種情況可能是對(duì)應(yīng)的時(shí)間段內(nèi)沒(méi)有流量請(qǐng)求到provider上
threadInfoBO.setQueueLength(0);
threadInfoBO.setActivePoolSize(0);
}
//這里是上報(bào)器上報(bào)數(shù)據(jù)到redis中
reportTemplate.reportData(JSON.toJSONString(threadInfoBO));
//上報(bào)之后，這里會(huì)重置map中的數(shù)據(jù)
DUBBO_INFO_STORE_CENTER.cleanInfo(DUBBO_PORT);
LOGGER.info(" =========== Dubbo線程池?cái)?shù)據(jù)上報(bào) =========== ");
} catch (Exception e) {
e.printStackTrace();
}
}
}
});
}
}
org.idea.dubbo.monitor.core.report.IReportTemplate
package org.idea.dubbo.monitor.core.report;
/**
* 上報(bào)模版
*
* @Author idea
* @Date created in 7:10 下午 2022/6/29
*/
public interface IReportTemplate {
/**
* 上報(bào)數(shù)據(jù)
*
* @return
*/
boolean reportData(String json);
}
package org.idea.dubbo.monitor.core.report.impl;
import org.idea.dubbo.monitor.core.report.IReportTemplate;
import org.idea.qiyu.cache.redis.service.IRedisService;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.time.LocalDate;
import java.util.concurrent.TimeUnit;
/**
* @Author idea
* @Date created in 7:12 下午 2022/6/29
*/
@Component
public class RedisTemplateImpl implements IReportTemplate {
@Resource
private IRedisService redisService;
private static String queueKey = "dubbo:threadpool:info:";
@Override
public boolean reportData(String json) {
redisService.lpush(queueKey + LocalDate.now().toString(), json);
redisService.expire(queueKey + LocalDate.now().toString(),7, TimeUnit.DAYS);
return true;
}
}
數(shù)據(jù)展示
推薦閱讀：
被呼吁抵制，7-Zip偽開(kāi)源還留有后門(mén)？
Lombok原理和同時(shí)使?@Data和@Builder 的坑
互聯(lián)網(wǎng)初中高級(jí)大廠面試題(9個(gè)G) 內(nèi)容包含Java基礎(chǔ)、JavaWeb、MySQL性能優(yōu)化、JVM、鎖、百萬(wàn)并發(fā)、消息隊(duì)列、高性能緩存、反射、Spring全家桶原理、微服務(wù)、Zookeeper......等技術(shù)棧！
?戳閱讀原文領(lǐng)??！ 朕已閱
評(píng)論
圖片
表情