目标
- 实现动态调整线程池参数
- 对线程池运行情况进行监控
- 核心线程数
- 超时时间
- 最大线程数
- 拒绝策略
文章图片
而队列BlockingQueue因为是final类型，所以没有对外修改入口。但可以通过重写LinkedBlockingQueue并把capacity设置为非final。
二，结合配置中心实现动态调整
这里的配置中心使用Apollo, 通过监听配置中心变化，然后更新线程池配置。示例代码如下：
@Slf4j
@Component
public class DynamicThreadPoolConfig {
/** 线程执行器 **/
private volatile ThreadPoolExecutor executor;
/** 核心线程数 **/
private Integer corePoolSize = 10;
/** 最大值线程数 **/
private Integer maximumPoolSize = 20;
/** 待执行任务的队列的长度 **/
private Integer workQueueSize = 1000;
/** 线程空闲时间 **/
private Long keepAliveTime = 1000L;
/** 线程名 **/
private String threadName;
private Config config = ConfigService.getConfig("lepu-activity-center");
;
public DynamicThreadPoolConfig() {
init(config);
}/** * 初始化 */
private void init(Config config) {
log.info("线程池初始化中..........");
if (executor == null) {
synchronized (DynamicThreadPoolConfig.class) {
if (executor == null) {
String corePoolSizeProperty = config.getProperty("corePoolSize", corePoolSize.toString());
log.info("修改前的核心线程池:{}",corePoolSizeProperty);
String maximumPoolSizeProperty = config.getProperty("maximumPoolSize", maximumPoolSize.toString());
String keepAliveTImeProperty = config.getProperty("keepAliveTime", keepAliveTime.toString());
BlockingQueue workQueueProperty = new LinkedBlockingQueue<>(workQueueSize);
executor = new ThreadPoolExecutor(Integer.valueOf(corePoolSizeProperty), Integer.valueOf(maximumPoolSizeProperty),
Long.valueOf(keepAliveTImeProperty), TimeUnit.MILLISECONDS, workQueueProperty);
}
}
}
}/**
* 监听到配置中心发生变化后，更新线程池配置
* @param changeEvent
*/
@ApolloConfigChangeListener
public void onChange(ConfigChangeEvent changeEvent){
log.info("线程池参数配置发生变化,namespace:{}",changeEvent.getNamespace());
for(String key : changeEvent.changedKeys()){
ConfigChange change = changeEvent.getChange(key);
String newValue = https://www.it610.com/article/change.getNewValue();
refreshThreadPool(key,newValue);
}
}/**
* 更新线程池配置
* @param key
* @param newValue
*/
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=https://www.it610.com/article/{}",key,newValue);
}
if (ParamsEnum.MAXIMUM_POOL_SIZE.getParam().equals(key)) {
executor.setMaximumPoolSize(Integer.valueOf(newValue));
log.info("修改最大线程数key={},value=https://www.it610.com/article/{}", key, newValue);
}
if (ParamsEnum.KEEP_ALIVE_TIME.getParam().equals(key)) {
executor.setKeepAliveTime(Integer.valueOf(newValue), TimeUnit.MILLISECONDS);
log.info("修改线程空闲时间key={},value=https://www.it610.com/article/{}", key, newValue);
}
}public ThreadPoolExecutor getExecutor() {
return executor;
}
} 三，监控方式
修改线程池有关参数重要，但知道何时修改同样重要，可以考虑间隔一段时间进行采集，通过日志输出，达到临界点后告警。
同样，ThreadPoolExecutor也提供获取线程池相关信息的API：
文章图片
【动态调整线程池参数】具体实现再看～
