什么是動(dòng)態(tài)線程池？

在線程池日常實(shí)踐中我們常常會(huì)遇到以下問題：

代碼中創(chuàng)建了一個(gè)線程池卻不知道核心參數(shù)設(shè)置多少比較合適。

參數(shù)設(shè)置好后，上線發(fā)現(xiàn)需要調(diào)整，改代碼重啟服務(wù)非常麻煩。

線程池相對于開發(fā)人員來說是個(gè)黑箱，運(yùn)行情況在出現(xiàn)問題 前很難被感知。

因此，動(dòng)態(tài)可監(jiān)控線程池一種針對以上痛點(diǎn)開發(fā)的線程池管理工具。主要可實(shí)現(xiàn)功能有：提供對 Spring 應(yīng)用內(nèi)線程池實(shí)例的全局管控、應(yīng)用運(yùn)行時(shí)動(dòng)態(tài)變更線程池參數(shù)以及線程池?cái)?shù)據(jù)采集和監(jiān)控閾值報(bào)警。

已經(jīng)實(shí)現(xiàn)的優(yōu)秀開源動(dòng)態(tài)線程池

hippo4j、dynamic-tp.....

實(shí)現(xiàn)思路

核心管理類

需要能實(shí)現(xiàn)對線程池的

服務(wù)注冊

獲取已經(jīng)注冊好的線程池

以及對注冊號線程池參數(shù)的刷新。

對于每一個(gè)線程池，我們使用一個(gè)線程池名字作為標(biāo)識(shí)每個(gè)線程池的唯一ID。

偽代碼實(shí)現(xiàn)

publicclassDtpRegistry{
/**
*儲(chǔ)存線程池
*/
privatestaticfinalMapEXECUTOR_MAP=newConcurrentHashMap<>();

/**
*獲取線程池
*@paramexecutorName線程池名字
*/
publicstaticExecutorgetExecutor(StringexecutorName){
returnEXECUTOR_MAP.get(executorName);
}

/**
*線程池注冊
*@paramexecutorName線程池名字
*/
publicstaticvoidregistry(StringexecutorName,Executorexecutor){
//注冊
EXECUTOR_MAP.put(executorName,executorWrapper);
}


/**
*刷新線程池參數(shù)
*@paramexecutorName線程池名字
*@paramproperties線程池參數(shù)
*/
publicstaticvoidrefresh(StringexecutorName,ThreadPoolPropertiesproperties){
Executorexecutor=EXECUTOR_MAP.get(executorName)
//刷新參數(shù)
//.......
}

}

如何創(chuàng)建線程池？

STEP 1. 我們可以使用yml配置文件的方式配置一個(gè)線程池，將線程池實(shí)例的創(chuàng)建交由Spring容器。

相關(guān)配置

publicclassDtpProperties{

privateListexecutors;

}

publicclassThreadPoolProperties{
/**
*標(biāo)識(shí)每個(gè)線程池的唯一名字
*/
privateStringpoolName;
privateStringpoolType="common";

/**
*是否為守護(hù)線程
*/
privatebooleanisDaemon=false;

/**
*以下都是核心參數(shù)
*/
privateintcorePoolSize=1;
privateintmaximumPoolSize=1;
privatelongkeepAliveTime;
privateTimeUnittimeUnit=TimeUnit.SECONDS;
privateStringqueueType="arrayBlockingQueue";
privateintqueueSize=5;
privateStringthreadFactoryPrefix="-td-";
privateStringRejectedExecutionHandler;
}

yml example:

spring:
dtp:
executors:
#線程池1
-poolName:dtpExecutor1
corePoolSize:5
maximumPoolSize:10
#線程池2
-poolName:dtpExecutor2
corePoolSize:2
maximumPoolSize:15

STEP 2 根據(jù)配置信息添加線程池的BeanDefinition

關(guān)鍵類

@Slf4j
publicclassDtpImportBeanDefinitionRegistrarimplementsImportBeanDefinitionRegistrar,EnvironmentAware{
privateEnvironmentenvironment;

@Override
publicvoidregisterBeanDefinitions(AnnotationMetadataimportingClassMetadata,BeanDefinitionRegistryregistry){
log.info("注冊");
//綁定資源
DtpPropertiesdtpProperties=newDtpProperties();
ResourceBundlerUtil.bind(environment,dtpProperties);
Listexecutors=dtpProperties.getExecutors();
if(Objects.isNull(executors)){
log.info("未檢測本地到配置文件線程池");
return;
}
//注冊beanDefinition
executors.forEach((executorProp)->{
BeanUtil.registerIfAbsent(registry,executorProp);
});
}


@Override
publicvoidsetEnvironment(Environmentenvironment){
this.environment=environment;
}
}


/**
*
*工具類
*
*/
publicclassBeanUtil{
publicstaticvoidregisterIfAbsent(BeanDefinitionRegistryregistry,ThreadPoolPropertiesexecutorProp){
register(registry,executorProp.getPoolName(),executorProp);
}

publicstaticvoidregister(BeanDefinitionRegistryregistry,StringbeanName,ThreadPoolPropertiesexecutorProp){
ClassexecutorType=ExecutorType.getClazz(executorProp.getPoolType());
Object[]args=assembleArgs(executorProp);
register(registry,beanName,executorType,args);
}

publicstaticvoidregister(BeanDefinitionRegistryregistry,StringbeanName,Classclazz,Object[]args){
BeanDefinitionBuilderbuilder=BeanDefinitionBuilder.genericBeanDefinition(clazz);
for(Objectarg:args){
builder.addConstructorArgValue(arg);
}
registry.registerBeanDefinition(beanName,builder.getBeanDefinition());
}

privatestaticObject[]assembleArgs(ThreadPoolPropertiesexecutorProp){
returnnewObject[]{
executorProp.getCorePoolSize(),
executorProp.getMaximumPoolSize(),
executorProp.getKeepAliveTime(),
executorProp.getTimeUnit(),
QueueType.getInstance(executorProp.getQueueType(),executorProp.getQueueSize()),
newNamedThreadFactory(
executorProp.getPoolName()+executorProp.getThreadFactoryPrefix(),
executorProp.isDaemon()
),
//先默認(rèn)不做設(shè)置
RejectPolicy.ABORT.getValue()
};
}
}

下面解釋一下這個(gè)類的作用，environment實(shí)例中儲(chǔ)存著spring啟動(dòng)時(shí)解析的yml配置，所以我們spring提供的Binder將配置綁定到我們前面定義的DtpProperties類中，方便后續(xù)使用。接下來的比較簡單，就是將線程池的BeanDefinition注冊到IOC容器中，讓spring去幫我們實(shí)例化這個(gè)bean。

STEP 3. 將已經(jīng)實(shí)例化的線程池注冊到核心類 DtpRegistry 中

我們注冊了 beanDefinition 后，spring會(huì)幫我們實(shí)例化出來， 在這之后我們可以根據(jù)需要將這個(gè)bean進(jìn)行進(jìn)一步的處理，spring也提供了很多機(jī)制讓我們對bean的生命周期管理進(jìn)行更多的擴(kuò)展。對應(yīng)到這里我們就是將實(shí)例化出來的線程池注冊到核心類 DtpRegistry 中進(jìn)行管理。

這里我們使用 BeanPostProcessor 進(jìn)行處理。

@Slf4j
publicclassDtpBeanPostProcessorimplementsBeanPostProcessor{
privateDefaultListableBeanFactorybeanFactory;

@Override
publicObjectpostProcessAfterInitialization(Objectbean,StringbeanName)throwsBeansException{
if(beaninstanceofDtpExecutor){
//直接納入管理
DtpRegistry.registry(beanName,(DtpExecutor)bean);
}
returnbean;
}
}

這里的邏輯很簡單， 就是判斷一下這個(gè)bean是不是線程池，是就統(tǒng)一管理起來。

STEP 4. 啟用 BeanDefinitionRegistrar 和 BeanPostProcessor

在springboot程序中，只要加一個(gè)@MapperScan注解就能啟用mybatis的功能，我們可以學(xué)習(xí)其在spring中的啟用方式，自定義一個(gè)注解：

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Import(DtpImportSelector.class)
public@interfaceEnableDynamicThreadPool{
}

其中，比較關(guān)鍵的是@Import注解，spring會(huì)導(dǎo)入注解中的類DtpImportSelector

而DtpImportSelector這個(gè)類實(shí)現(xiàn)了：

publicclassDtpImportSelectorimplementsDeferredImportSelector{
@Override
publicString[]selectImports(AnnotationMetadataimportingClassMetadata){
returnnewString[]{
DtpImportBeanDefinitionRegistrar.class.getName(),
DtpBeanPostProcessor.class.getName()
};
}
}

這樣，只要我們再啟動(dòng)類或者配置類上加上@EnableDynamicThreadPool這個(gè)注解，spring就會(huì)將DtpImportBeanDefinitionRegistrar和DtpBeanPostProcessor這兩個(gè)類加入spring容器管理，從而實(shí)現(xiàn)我們的線程池的注冊。

@SpringBootApplication
@EnableDynamicThreadPool
publicclassApplication{
publicstaticvoidmain(String[]args){
SpringApplication.run(Application.class,args);
}
}

如何實(shí)現(xiàn)線程池配置的動(dòng)態(tài)刷新

首先明確一點(diǎn)，對于線程池的實(shí)現(xiàn)類，例如：ThreadPoolExecutor等，都有提供核心參數(shù)對應(yīng)的 set 方法，讓我們實(shí)現(xiàn)參數(shù)修改。因此，在核心類DtpRegistry中的refresh方法，我們可以這樣寫：

publicclassDtpRegistry{
/**
*儲(chǔ)存線程池
*/
privatestaticfinalMapEXECUTOR_MAP=newConcurrentHashMap<>();
/**
*刷新線程池參數(shù)
*@paramexecutorName線程池名字
*@paramproperties線程池參數(shù)
*/
publicstaticvoidrefresh(StringexecutorName,ThreadPoolPropertiesproperties){
ThreadPoolExecutorexecutor=EXECUTOR_MAP.get(executorName)

//設(shè)置參數(shù)
executor.setCorePoolSize(...);
executor.setMaximumPoolSize(...);
......
}

}

而這些新參數(shù)怎么來呢？我們可以引入Nacos、Apollo等配置中心，實(shí)現(xiàn)他們的監(jiān)聽器方法，在監(jiān)聽器方法里調(diào)用DtpRegistry的refresh方法刷新即可。

審核編輯：劉清