动态线程池思想学习与实践

在后台项目开发过程中，我们常常借助线程池来实现多线程任务，以此提升系统的吞吐率和响应性；而线程池的参数配置却是一个难以合理评估的值，虽然业界也针对CPU密集型，IO密集型等场景给出了一些参数配置的经验与方案，但是实际业务场景中通常会因为流量的随机性，业务的更迭性等情况出现预计和实际运行情况偏差较大的情况；而不合理的线程池参数，可能导致服务器负载升高，服务不可用，内存溢出等严重问题；一旦遇到参数不合理的问题，还需要重新上线修改，并且存在反复修改的情况，而这期间花费的时间可能带来更大的风险，甚至导致严重业务事故；那么有没有一种方式能有效感知上述问题并及时避免以上问题呢？或许动态线程池可以。

public class ThreadPoolManager {    // 线程池管理器    private static final ConcurrentHashMap<String, Executor> REGISTER_MAP_BY_NAME = new ConcurrentHashMap<>();    private static final ConcurrentHashMap<Executor, String> REGISTER_MAP_BY_EXECUTOR = new ConcurrentHashMap<>();
    // 注册线程池     public static void registerExecutor(String threadPoolName, Executor executor) {        REGISTER_MAP_BY_NAME.putIfAbsent(threadPoolName, executor);        REGISTER_MAP_BY_EXECUTOR.putIfAbsent(executor, threadPoolName);    }
    // 根据名称获取线程池    public static Executor getExecutorByName(String threadPoolName) {        return REGISTER_MAP_BY_NAME.get(threadPoolName);    }
    // 根据线程池获取名称    public static String getNameByExecutor(Executor executor) {        return REGISTER_MAP_BY_EXECUTOR.get(executor);    }    // 获取所有线程池名称    public static Set<String> getAllExecutorNames() {        return REGISTER_MAP_BY_NAME.keySet();    }}

getCorePoolSize() // 核心线程数getMaximumPoolSize() // 最大线程数getQueue() // 阻塞队列，获取队列大小，容量等getActiveCount() // 活跃线程数getTaskCount() // 历史已完成和正在执行的任务数量getCompletedTaskCount() // 已完成任务数

// 已经设置埋点的线程池public static ConcurrentHashSet<String> monitorThreadPool = new ConcurrentHashSet<>();
// 监控埋点注册public static void monitorRegister() {    log.info("===> monitor register start...");    // 1.获取所有线程池    Set<String> allExecutorNames = ThreadPoolManager.getAllExecutorNames();    // 2.遍历线程池，注册埋点    allExecutorNames.forEach(executorName-> {        if (!monitorThreadPool.contains(executorName)) {            monitorThreadPool.add(executorName);            Executor executor = ThreadPoolManager.getExecutorByName(executorName);            collect(executor, executorName);        }    });    log.info("===> monitor register end...");}
// pfinder指标埋点public static void collect(Executor executorService, String threadPoolName) {    ThreadPoolExecutor executor = (ThreadPoolExecutor)executorService;    String prefix = "thread.pool."+threadPoolName;    gauge1 = PfinderContext.getMetricRegistry().gauges(prefix)            .gauge(() -> executor.isShutdown() ? 0 : executor.getCorePoolSize())            .tags(MetricTag.of("type_dimension", "core_size")).build();    gauge2 = PfinderContext.getMetricRegistry().gauges(prefix)            .gauge(() -> executor.isShutdown() ? 0 : executor.getMaximumPoolSize())            .tags(MetricTag.of("type_dimension", "max_size"))            .build();    gauge4 = PfinderContext.getMetricRegistry().gauges(prefix)            .gauge(() -> executor.isShutdown() ? 0 : executor.getQueue().size())            .tags(MetricTag.of("type_dimension", "queue_size"))            .build();}

@Slf4jpublic class RejectInvocationHandler implements InvocationHandler {
    private final Object target;
    @Value("${jtool.pool.reject.alarm.key}")    private String key;
    public RejectInvocationHandler(Object target) {        this.target = target;    }
    @Override    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {        ExecutorService executor = (ExecutorService)args[1];        if (Strings.equals(method.getName(), "rejectedExecution")) {            try {                rejectBefore(executor);            }            catch (Exception exp) {                log.error("==> Exception while do rejectBefore for pool [{}]", executor, exp);            }        }        return method.invoke(target, args);    }
    private void rejectBefore(ExecutorService executor) {        // 触发报警          rejectAlarm(executor);     }
    /**     * 拒绝报警     */    private void rejectAlarm(ExecutorService executor) {        String alarmKey = Objects.nonNull(key) ? key : ThreadPoolConst.UMP_ALARM_KEY;        ThreadPoolExecutor threadPoolExecutor = (ThreadPoolExecutor)executor;        String threadPoolName = ThreadPoolManager.getNameByExecutor(threadPoolExecutor);        String errorMsg = String.format("===> 线程池拒绝报警 key: [%s], cur executor: [%s], core size: [%s], max size: [%s], queue size: [%s], curQueue size: [%s]",                alarmKey, threadPoolName, threadPoolExecutor.getCorePoolSize(), threadPoolExecutor.getMaximumPoolSize(), threadPoolExecutor.getQueue().size()+threadPoolExecutor.getQueue().remainingCapacity(), threadPoolExecutor.getQueue().size());        log.error(errorMsg);        Profiler.businessAlarm(alarmKey, errorMsg);    }
}

public class RejectInvocationHandler implements InvocationHandler {...private void rejectBefore(ExecutorService executor) {        // 打印线程堆栈到日志的间隔条件        if (CommonProperty.canPrintStackTrace()) {             // 触发报警             rejectAlarm(executor);             // 触发线程堆栈打印             printThreadStack(executor);         }    }...}
.../** * 打印线程堆栈信息 */public static void printThreadStack(Executor executor) {    if (!CommonProperty.logPrintFlag) {        log.info("===> 线程池堆栈打印关闭：[{}]", CommonProperty.logPrintFlag);        return;    }    logger.info("\n=================>>> 线程池拒绝堆栈打印start，触发提交拒接处理的线程池：【{}】", executor);    Map<Thread, StackTraceElement[]> allStackTraces = Thread.getAllStackTraces();    log.info("===> allStackTraces size :[{}]", allStackTraces.size());    StringBuilder stringBuilder = new StringBuilder();    allStackTraces.entrySet().stream()            .sorted(Comparator.comparing(entry -> entry.getKey().getName()))            .forEach(threadEntry -> {                Thread thread = threadEntry.getKey();                stringBuilder.append(Strings.format("\n线程：[{}] 时间: [{}]\n", thread.getName(), new Date()));                stringBuilder.append(Strings.format("\tjava.lang.Thread.State: {}\n", thread.getState()));                StackTraceElement[] stack = threadEntry.getValue();                for (StackTraceElement stackTraceElement : stack) {                    stringBuilder.append("\t\t").append(stackTraceElement.toString()).append("\n");                }                stringBuilder.append("\n");                logger.info(stringBuilder.toString());                stringBuilder.delete(0, stringBuilder.length());            });    logger.info("==============>>> end");}...

public class ThreadLogAnalyzer {    public static void main(String[] args) {        String logFilePath = "/Users/demo/jsfdemo/MyJtool/src/main/resources/reject.monitor 21.log";        String threadPoolNameLike = "simpleTestExecutor";        int threadCount = 0;        HashMap<String, Integer> statusMap = new HashMap<>();        HashMap<String, Integer> methodMap = new HashMap<>();
        try (BufferedReader br = new BufferedReader(new FileReader(logFilePath))) {            String line;            while ((line = br.readLine()) != null) {                if (line.contains("=================>>> 线程池拒绝堆栈打印start，触发提交拒接处理的线程池")) {                    System.out.println("开始读整个线程");                }                if (line.contains(threadPoolNameLike)) {                    threadCount++;                    String curStatus = br.readLine();                    if (curStatus.contains("java.lang.Thread.State")) {                        statusMap.put(curStatus, (statusMap.getOrDefault(curStatus, 0) + 1));                        String methodTrace = br.readLine();                        methodMap.put(methodTrace, (methodMap.getOrDefault(methodTrace, 0) + 1));                    }                }                if (line.contains("==============>>> end")) {                    System.out.println("结束读整个线程");                }            }        } catch (IOException e) {            e.printStackTrace();        }
        System.out.println(Strings.format("===> 当前线程名[{}]共计：{}", threadPoolNameLike, threadCount));        System.out.println("\n===> 状态分析结果：");        for (Map.Entry<String, Integer> statusEntry : statusMap.entrySet()) {            System.out.println(Strings.format("\t {} {}", statusEntry.getKey(), statusEntry.getValue()));        }        System.out.println("\n===> 方法分析结果：");        ArrayList<Map.Entry<String, Integer>> methodEntryList = Lists.newArrayList(methodMap.entrySet());        methodEntryList.sort(new Comparator<Map.Entry<String, Integer>>() {            @Override            public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {                return o2.getValue() - o1.getValue();            }        });        for (Map.Entry<String, Integer> methodEntry : methodEntryList) {            System.out.println(Strings.format("{} {} {}%", methodEntry.getKey(), methodEntry.getValue(), (methodEntry.getValue() / (double)threadCount) * 100));        }    }}

/** * ducc控制线程池刷新方法, 需要动态刷新的线程池信息列表，举例如下： * value: * [ *   { *     "threadPoolName": "my_pool", *     "corePoolSize": "10", *     "maximumPoolSize": "20", *     "queueCapacity": "100" *   } * ] */@LafValue("jtool.pool.refresh")public void refresh(@JsonConverter List<ThreadPoolProperties> threadPoolProperties) {    String jsonString = JSON.toJSONString(threadPoolProperties);    log.info("===> refresh thread pool properties [{}]", jsonString);    threadPoolProperties = JSONObject.parseArray(jsonString, ThreadPoolProperties.class);    refresh(threadPoolProperties);}
public static boolean refresh(List<ThreadPoolProperties> threadPoolProperties) {    if (Objects.isNull(threadPoolProperties)) {        log.warn("refresh param is empty!");        return false;    }    log.info("Executor refresh param: [{}]", threadPoolProperties);    // 1.根据参数获取对应的线程池    threadPoolProperties.forEach(threadPoolProperty -> {        String threadPoolName = threadPoolProperty.getThreadPoolName();        Executor executor = ThreadPoolManager.getExecutorByName(threadPoolName);        if (Objects.isNull(executor)) {            log.warn("Register not find this executor: {}", threadPoolName);            return;        }        // 2. 线程池刷新        refreshExecutor(executor, threadPoolName, threadPoolProperty);        log.info("Refresh thread pool finish, threadPoolName: [{}]", threadPoolName);    });    return true;}