Java实现监控Dubbo线程池

布满荆棘的人生 2023-01-02 01:28 107阅读 0赞

alibaba dubbo和apache dubbo 获取线程池的方式是不一样的

alibaba dubbo线程池监控

import com.alibaba.dubbo.common.Constants;
    import com.alibaba.dubbo.common.extension.ExtensionLoader;
    import com.alibaba.dubbo.common.store.DataStore;
    import com.google.common.collect.Maps;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.factory.InitializingBean;
     
    import java.net.InetAddress;
    import java.util.Map;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.ThreadPoolExecutor;
     
    /** * dubbo线程池监控 * * @author huchenfei * @version 1.0 * @class DubboThreadPoolMonitorTask * @date 2020/12/28 18:03 * @description */
    public class DubboThreadPoolMonitorJob { 
     
        protected static final Logger LOCAL_LOGGER = LoggerFactory.getLogger(DubboThreadPoolMonitorJob.class);
     
        private static final String MONITOR_PREFIX = "dubbo_threadPool.";
     
        private static final Map<String, Long> TASK_COUNT_MAP = Maps.newConcurrentMap();
     
        public void execute() { 
            try { 
                DataStore dataStore = ExtensionLoader.getExtensionLoader(DataStore.class).getDefaultExtension();
                if (dataStore == null) { 
                    return;
                }
     
                InetAddress localHost = InetAddress.getLocalHost();
                String hostName = localHost.getHostName().replace('.', '_');
                String host = hostName + ":";
                Map<String, Object> executors = dataStore.get(Constants.EXECUTOR_SERVICE_COMPONENT_KEY);
                for (Map.Entry<String, Object> entry : executors.entrySet()) { 
                    // Dubbo的端口
                    String port = entry.getKey();
                    if (!"20880".equals(port)) { 
                        continue;
                    }
                    // Dubbo线程池,一个端口对应一个线程池
                    ExecutorService executor = (ExecutorService) entry.getValue();
     
                    if (!(executor instanceof ThreadPoolExecutor)) { 
                        continue;
                    }
     
                    // Dubbo线程池的监控Key
                    String monitorPort = port + ".";
                    // dubbo_threadPool.host.monitorKey
                    String prefix = MONITOR_PREFIX + host + monitorPort;
                    ThreadPoolExecutor tp = (ThreadPoolExecutor) executor;
                    LOCAL_LOGGER.info("activeCount当前活跃线程数：" + tp.getActiveCount());
                    LOCAL_LOGGER.info("poolSize线程池size：" + tp.getPoolSize());
                    LOCAL_LOGGER.info("currentQueueSize当前排队线程数：" + tp.getQueue().size());
                    LOCAL_LOGGER.info("corePollSize核心线程数：" + tp.getCorePoolSize());
                    LOCAL_LOGGER.info("maximumPoolSize最大线程数：" + tp.getMaximumPoolSize());
     
                    String taskCountKey = prefix + "taskCount";
                    long lastTaskCount = TASK_COUNT_MAP.getOrDefault(taskCountKey, 0L);
                    TASK_COUNT_MAP.put(taskCountKey, tp.getTaskCount());
                    LOCAL_LOGGER.info("taskCount当前总线程数：" + Math.max(tp.getTaskCount() - lastTaskCount, 0L));
     
                    String completeCountKey = prefix + "completedTaskCount";
                    long lastCompleteCount = TASK_COUNT_MAP.getOrDefault(completeCountKey, 0L);
                    TASK_COUNT_MAP.put(completeCountKey, tp.getCompletedTaskCount());
                    LOCAL_LOGGER.info("completedTaskCount当前执行完成线程数：" + Math.max(tp.getCompletedTaskCount() - lastCompleteCount, 0L));
                }
            } catch (Exception e) { 
                LOCAL_LOGGER.info("record dubbo thread pool exception", e);
            }
        }
    }

apache dubbo线程监控

import com.google.common.collect.Maps;
    import org.apache.dubbo.common.constants.CommonConstants;
    import org.apache.dubbo.common.extension.ExtensionLoader;
    import org.apache.dubbo.common.threadpool.manager.DefaultExecutorRepository;
    import org.apache.dubbo.common.threadpool.manager.ExecutorRepository;
    import org.slf4j.Logger;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.scheduling.annotation.EnableScheduling;
    import org.springframework.scheduling.annotation.Scheduled;
    import org.springframework.stereotype.Service;
    
    import java.net.InetAddress;
    import java.util.Map;
    import java.util.concurrent.ConcurrentMap;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.ThreadPoolExecutor;
    
    /** * dubbo线程池监控 * * @author huchenfei * @version 1.0 * @class DubboThreadPoolMonitorTask * @date 2020/12/28 18:03 * @description */
    @EnableScheduling
    @Service
    public class DubboThreadPoolMonitorJob { 
    
    	private static final Logger logger = LoggerFactory.getLogger(DubboThreadPoolMonitorJob.class);
        
        private static final String MONITOR_PREFIX = "dubbo_threadPool.";
    
        private static final Map<String, Long> TASK_COUNT_MAP = Maps.newConcurrentMap();
    
        @Scheduled(cron="0 0/1 * * * ?")
        public void execute() { 
            try { 
                ExecutorRepository executorRepository = ExtensionLoader.getExtensionLoader(ExecutorRepository.class).getDefaultExtension();
    
                if (executorRepository instanceof DefaultExecutorRepository) { 
                    DefaultExecutorRepository defaultExecutorRepository = (DefaultExecutorRepository) executorRepository;
                    ConcurrentMap<String, ConcurrentMap<Integer, ExecutorService>> data = (ConcurrentMap<String, ConcurrentMap<Integer, ExecutorService>>) ReflectUtil.read(defaultExecutorRepository, "data");
    
                    if (data == null) { 
                        return;
                    }
                    //provider
                    ConcurrentMap<Integer, ExecutorService> executors = data.get(CommonConstants.EXECUTOR_SERVICE_COMPONENT_KEY);
                    if (executors == null) { 
                        return;
                    }
                    InetAddress localHost = InetAddress.getLocalHost();
                    String hostName = localHost.getHostName().replace('.', '_');
                    String host = hostName + ":";
                    for (Map.Entry<Integer, ExecutorService> entry : executors.entrySet()) { 
                        // Dubbo的端口
                        int port = entry.getKey();
                        if (20880 != port) { 
                            continue;
                        }
                        // Dubbo线程池,一个端口对应一个线程池
                        ExecutorService executor = entry.getValue();
                        if (!(executor instanceof ThreadPoolExecutor)) { 
                            continue;
                        }
                        // Dubbo线程池的监控Key
                        String monitorPort = port + ".";
                        // dubbo_threadPool.host.monitorKey
                        String prefix = MONITOR_PREFIX + host + monitorPort;
                        ThreadPoolExecutor tp = (ThreadPoolExecutor) executor;
    
                        logger.info(prefix + "activeCount当前活跃线程数：" + tp.getActiveCount());
                        logger.info(prefix + "poolSize线程池size：" + tp.getPoolSize());
                        logger.info(prefix + "currentQueueSize当前排队线程数：" + tp.getQueue().size());
                        logger.info(prefix + "corePollSize核心线程数：" + tp.getCorePoolSize());
                        logger.info(prefix + "maximumPoolSize最大线程数：" + tp.getMaximumPoolSize());
    
                        String taskCountKey = prefix + "taskCount";
                        long lastTaskCount = TASK_COUNT_MAP.getOrDefault(taskCountKey, 0L);
                        TASK_COUNT_MAP.put(taskCountKey, tp.getTaskCount());
                        logger.info(prefix + "taskCount当前总线程数：" + Math.max(tp.getTaskCount() - lastTaskCount, 0L));
    
                        String completeCountKey = prefix + "completedTaskCount";
                        long lastCompleteCount = TASK_COUNT_MAP.getOrDefault(completeCountKey, 0L);
                        TASK_COUNT_MAP.put(completeCountKey, tp.getCompletedTaskCount());
                        logger.info(prefix + "completedTaskCount当前执行完成线程数：" + Math.max(tp.getCompletedTaskCount() - lastCompleteCount, 0L));
                    }
                }
            } catch (Exception e) { 
                logger.error("record dubbo thread pool exception", e);
            }
        }
    }
    
    
    import org.apache.commons.lang3.reflect.FieldUtils;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    /** * 反射类 * * @author huchenfei * @version 1.0· * @class ReflectUtil * @date 2020/12/30 20:37 * @description */
    public class ReflectUtil { 
    
        private static final Logger log = LoggerFactory.getLogger(ReflectUtil.class);
    
        private ReflectUtil() { 
        }
    
        public static Object read(final Object target, final String fieldName) { 
            try { 
                return FieldUtils.readField(target, fieldName, true);
            } catch (IllegalAccessException var3) { 
                log.error("读取值失败", var3);
                return null;
            }
        }
    }