动态调整线程池参数实践

<dependencies>
    <dependency>
        <groupId>com.ctrip.framework.apollo</groupId>
        <artifactId>apollo-client</artifactId>
        <version>1.7.0</version>
    </dependency>
    <dependency>
        <groupId>org.slf4j</groupId>
        <artifactId>slf4j-simple</artifactId>
        <version>1.7.25</version>
        <scope>compile</scope>
    </dependency>
</dependencies>

/**
 * 动态获取Apollo配置
 * 注意要配置：-Dapp.id=myApp -Denv=DEV -Dapollo.cluster=default -Ddev_meta=http://localhost:8080
 *
 * @author 微信公众号【老周聊架构】
 */
public class GetApolloConfigTest {
    public static void main(String[] args) throws InterruptedException {
        Config config = ConfigService.getAppConfig();
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                System.out.println("Changes for namespace " + changeEvent.getNamespace());
                for (String key : changeEvent.changedKeys()) {
                    ConfigChange change = changeEvent.getChange(key);
                    System.out.println(String.format("Found change - key: %s, oldValue: %s, newValue: %s, changeType: %s", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType()));
                }
            }
        });
        Thread.sleep(1000000L);
    }
}

/**
 * 动态线程池工厂
 *
 * @author 微信公众号【老周聊架构】
 */
@Slf4j
@Component
public class DynamicThreadPoolFactory {
    /** 这里是你的namespace,我这里是默认的application **/
    private static final String NAME_SPACE = "application";
    /** 线程执行器 **/
    private volatile ThreadPoolExecutor executor;
    /** 核心线程数 **/
    private Integer corePoolSize = 10;
    /** 最大值线程数 **/
    private Integer maximumPoolSize = 20;
    /** 待执行任务的队列的长度 **/
    private Integer workQueueSize = 1000;
    /** 线程空闲时间 **/
    private Long keepAliveTime = 1000L;
    /** 线程名 **/
    private String threadName;
    public DynamicThreadPoolFactory() {
        Config config = ConfigService.getConfig(NAME_SPACE);
        init(config);
        listen(config);
    }
    /**
     * 初始化
     */
    private void init(Config config) {
        if (executor == null) {
            synchronized (DynamicThreadPoolFactory.class) {
                if (executor == null) {
                    String corePoolSizeProperty = config.getProperty(ParamsEnum.CORE_POOL_SIZE.getParam(), corePoolSize.toString());
                    String maximumPoolSizeProperty = config.getProperty(ParamsEnum.MAXIMUM_POOL_SIZE.getParam(), maximumPoolSize.toString());
                    String keepAliveTImeProperty = config.getProperty(ParamsEnum.KEEP_ALIVE_TIME.getParam(), keepAliveTime.toString());
                    BlockingQueue<Runnable> workQueueProperty = new LinkedBlockingQueue<>(workQueueSize);
                    executor = new ThreadPoolExecutor(Integer.valueOf(corePoolSizeProperty), Integer.valueOf(maximumPoolSizeProperty),
                            Long.valueOf(keepAliveTImeProperty), TimeUnit.MILLISECONDS, workQueueProperty);
                }
            }
        }
    }
    /**
     * 监听器
     */
    private void listen(Config config) {
        config.addChangeListener(new ConfigChangeListener() {
            @Override
            public void onChange(ConfigChangeEvent changeEvent) {
                log.info("命名空间发生变化={}", changeEvent.getNamespace());
                for (String key : changeEvent.changedKeys()) {
                    ConfigChange change = changeEvent.getChange(key);
                    String newValue = change.getNewValue();
                    refreshThreadPool(key, newValue);
                    log.info("发生变化key={},oldValue={},newValue={},changeType={}", change.getPropertyName(), change.getOldValue(), change.getNewValue(), change.getChangeType());
                }
            }
        });
    }
    /**
     * 刷新线程池
     */
    private void refreshThreadPool(String key, String newValue) {
        if (executor == null) {
            return;
        }
        if (ParamsEnum.CORE_POOL_SIZE.getParam().equals(key)) {
            executor.setCorePoolSize(Integer.valueOf(newValue));
            log.info("修改核心线程数key={},value={}", key, newValue);
        }
        if (ParamsEnum.MAXIMUM_POOL_SIZE.getParam().equals(key)) {
            executor.setMaximumPoolSize(Integer.valueOf(newValue));
            log.info("修改最大线程数key={},value={}", key, newValue);
        }
        if (ParamsEnum.KEEP_ALIVE_TIME.getParam().equals(key)) {
            executor.setKeepAliveTime(Integer.valueOf(newValue), TimeUnit.MILLISECONDS);
            log.info("修改线程空闲时间key={},value={}", key, newValue);
        }
    }
    public ThreadPoolExecutor getExecutor(String threadName) {
        return executor;
    }
}

@AllArgsConstructor
public enum ParamsEnum {
    CORE_POOL_SIZE("corePoolSize", "核心线程数"),
    MAXIMUM_POOL_SIZE("maximumPoolSize", "最大线程数"),
    KEEP_ALIVE_TIME("keepAliveTime", "线程空闲时间"),
    ;
    @Getter
    private String param;
    @Getter
    private String desc;
}

/**
 * 动态线程池执行器
 *  
 * @author 微信公众号【老周聊架构】
 */
@Component
public class DynamicThreadExecutor {
    @Resource
    private DynamicThreadPoolFactory threadPoolFactory;
    public void execute(String bizName, Runnable job) {
        threadPoolFactory.getExecutor(bizName).execute(job);
    }
    public Future<?> sumbit(String bizName, Runnable job) {
        return threadPoolFactory.getExecutor(bizName).submit(job);
    }
}

@Slf4j
public class DynamicThreadPoolExecutorTest {
    @Resource
    private DynamicThreadExecutor dynamicThreadExecutor;
    /**
     * 记得 IDEA VM options 要记得加下面的参数
     * -Dapp.id=SampleApp -Denv=DEV -Dapollo.meta=http://localhost:8080
     */
    @Test
    public void testExecute() throws InterruptedException {
        while (true) {
            dynamicThreadExecutor.execute("bizName", new Runnable() {
                @Override
                public void run() {
                    System.out.println("bizInfo");
                }
            });
            TimeUnit.SECONDS.sleep(1);
        }
    }
}

private static void threadPoolStatus(ThreadPoolExecutor executor, String name) {
    BlockingQueue<Runnable> queue = executor.getQueue();
    System.out.println(Thread.currentThread().getName() + "-" + name + "-:" +
            "核心线程数：" + executor.getCorePoolSize() +
            " 活动线程数：" + executor.getActiveCount() +
            " 最大线程数：" + executor.getMaximumPoolSize() +
            " 线程池活跃度：" + divide(executor.getActiveCount(), executor.getMaximumPoolSize()) +
            " 任务完成数：" + executor.getCompletedTaskCount() +
            " 队列大小：" + (queue.size() + queue.remainingCapacity()) +
            " 当前排队线程数：" + queue.size() +
            " 队列剩余大小：" + queue.remainingCapacity() +
            " 队列使用度：" + divide(queue.size(), queue.size() + queue.remainingCapacity()));
}