手把手教你java线程池动态配置调整

解释一下什么是线程池的动态调整？

java程序中使用线程池，在运行过程中动态的调整核心线程数（core_size）、最大线程数（max_size）、存放排队任务的queue大小（blockingQueue size）

为什么会来实现线程池的动态调整？

来自于美团的公众号“Java线程池实现原理及其在美团业务中的实践”中提到的动态化线程池，看到后眼前一亮，还有这种操作？为什么以前没想到过？所有的技术的方案都是业务所遇到的问题而产生出来的，我想美团应该是流量的不均衡，导致有时需要快速扩充处理能力，而不用发布代码产生的一种解决方案吧。

说完背景，我们就直接动手来实现吧

1，要实现运行时动态、那么需要动态的参数需要暴露set方法。美团的文章已经交代了，JDK原生线程池ThreadPoolExecutor提供了如下几个public的setter方法，如下图所示：

2，即然支持运行时set，那岂不是很简单，我们只需要选择一个可以热更新的组件就可以完成，线程池的动态化，这里我们选择配置中心apollo来作为热更新的组件。

初始化线程池（核心线程数=2，最大线程数=4，等待任务最大为10）

java复制代码 static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(2, 4,
             10L, TimeUnit.SECONDS,
             new CLinkedBlockingQueue<>(10),
             new ThreadFactoryBuilder().setNameFormat("c_t_%d").build(), new RejectedExecutionHandler() {
         @Override
         public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
             System.out.println("rejected");
         }
     });

每1s添加一个任务，每个任务执行时间为10S的话，这个线程池的变化应该是什么样的，如下图：

java复制代码public static void main(String[] args)  {

        initApolloConfig();

        Config config = ConfigService.getConfig("application");
        //apollo 值发生变更添加监听器
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent configChangeEvent) {
                ConfigChange threadCoreSize = configChangeEvent.getChange("thread_core_size");
                ConfigChange threadMaxSize = configChangeEvent.getChange("thread_max_size");
                ConfigChange queueSize = configChangeEvent.getChange("queue_size");
                if (threadCoreSize!=null){
                    threadPoolExecutor.setCorePoolSize(Integer.parseInt(threadCoreSize.getNewValue()));
                }
                if (threadMaxSize!=null){
                    threadPoolExecutor.setMaximumPoolSize(Integer.parseInt(threadMaxSize.getNewValue()));
                }
                if (queueSize!=null){
                    CLinkedBlockingQueue cLinkedBlockingQueue =(CLinkedBlockingQueue)threadPoolExecutor.getQueue();
                    cLinkedBlockingQueue.setCapacity(Integer.parseInt(queueSize.getNewValue()));
                }
            }
        });
        
        System.out.println("threadPoolExecutor init status:");
        printThreadPoolStatus();
        
        //每1S添加一个任务
        for (int i=0;i<100;i++){
            try {
                Thread.sleep(1000);
                dynamicThreadPoolAddTask(i);
                printThreadPoolStatus();
            }catch (InterruptedException ex){
                ex.printStackTrace();
            }
        }
    }
     /***
     * 打印当前线程池的状态
     */
    private static void  printThreadPoolStatus(){
        String s=String.format("core_size:%s,thread_current_size:%s;" +
                "thread_max_size:%s;queue_current_size:%s,total_task_count:%s",threadPoolExecutor.getCorePoolSize(),
                threadPoolExecutor.getActiveCount(),threadPoolExecutor.getMaximumPoolSize(),threadPoolExecutor.getQueue().size(),
                threadPoolExecutor.getTaskCount());
        System.out.println(s);
    }
     /***
     * 给线程池添加任务
     * @param i
     */
    private static void dynamicThreadPoolAddTask(int i){
        threadPoolExecutor.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    Thread.sleep(10000);
                    System.out.println(i);
                }catch (InterruptedException ex){

                }
            }
        });
    }