利用Nacos作为配置中心动态修改线程池核心参数

背景

这篇文章的主要核心原理都来自于这个开源项目dynamic-tp&＃xff0c;可以说是对这个开源项目的源码分析&＃xff0c;也是对这个开源项目中涉及到的技术点进行学习总结。

从这篇文章中能学到的技术点

从这篇文章中能学到的技术点&＃xff0c;也就是从这个dynamic-tp开源项目中学习到的技术点&＃xff08;这里只列举了这个项目的冰山一角&＃xff09;&＃xff1a;

• 利用Nacos作为配置中心&＃xff0c;动态监听Nacos配置变更。
• 项目启动之初获取Nacos配置&＃xff0c;将yml文件中配置的线程池注册到IOC容器中。这里在文章的最后再介绍&＃xff0c;利用Spring提供的扩展点。
• .yml文件映射到具体的对象中&＃xff0c;对比是否有变更。
• Spring提供的扩展点学习。

那我们开始吧

以下都是对dynamic-tp这个开源项目进行了简化&＃xff0c;首先看一下我的Nacos配置以及配置类&＃xff1a;

spring:dynamic:tp:enabled: trueexecutors:- threadPoolName: commonExecutorcorePoolSize: 2maximumPoolSize: 8queueCapacity: 200keepAliveTime: 50allowCoreThreadTimeOut: false- threadPoolName: dynamicThreadPoolExecutor1corePoolSize: 4maximumPoolSize: 6queueCapacity: 400keepAliveTime: 50allowCoreThreadTimeOut: false

&＃64;Data&＃64;ConfigurationProperties(prefix &＃61; "spring.dynamic.tp")public class DynamicThreadPoolProperties {​/*** 是否开始动态配置*/private boolean enabled &＃61; true;​/*** 线程池基础配置*/private List executors;​}

&＃64;Datapublic class ThreadPoolProperties {​/*** 线程池的名称*/private String threadPoolName;​/*** 核心线程数*/private int corePoolSize &＃61; 1;​/*** 最大线程数*/private int maximumPoolSize &＃61; DynamicThreadPoolConst.AVAILABLE_PROCESSORS;​/*** 线程队列数*/private int queueCapacity &＃61; 1024;​/*** 是否允许核心线程超时*/private boolean allowCoreThreadTimeOut &＃61; false;​/*** 超时时间*/private long keepAliveTime &＃61; 30;​/*** Timeout unit.*/private TimeUnit timeUnit &＃61; TimeUnit.SECONDS;}

上面这个技术点就不用多说了吧&＃xff0c;yml文件配置的内容会映射到DynamicThreadPoolProperties类中,多个的话会映射成集合的形式。

介绍完了基础的配置&＃xff0c;那我们开始介绍核心一点东西&＃xff1a;监听Nacos配置变更的事件。

监听Nacos配置变更的方式有多种&＃xff0c;可以使用实现 ApplicationListener的方式也可以使用注解&＃64;NacosConfigListener。

下面我们就使用实现 ApplicationListener的方式&＃xff1a;

&＃64;Componentpublic class NacosRefresher extends AbstractRefresher implements ApplicationListener {​&＃64;Overridepublic void onApplicationEvent(NacosConfigReceivedEvent nacosConfigReceivedEvent) {refresher(nacosConfigReceivedEvent.getContent());}}

当Nacos的配置发生了变更的时候会执行onApplicationEvent方法&＃xff1b;

AbstractRefresher类提供了refresher方法&＃xff0c;至于为啥做成一个抽象的是因为不光Nacos可以作为配置中心&＃xff0c;zookeeper也可以。像dynamic-tp作者不光提供了Nacos,Zookeeper,还有Apollo。

上面提到了当Nacos的配置发生了变更的时候会执行onApplicationEvent方法&＃xff0c;onApplicationEvent方法中调用了AbstractRefresher类提供的refresher方法:

&＃64;Slf4jpublic abstract class AbstractRefresher {​&＃64;Resourceprivate DynamicThreadPoolRegistry dynamicThreadPoolRegistry;​//项目启动之初就与.yml文件中的配置项进行映射了&＃64;Resourceprivate DynamicThreadPoolProperties dtpProperties;​public void refresher(String content) {if (StringUtils.isBlank(content)) {log.warn("DynamicTp refresh, empty content.");return;}//目前先支持yml解析YamlConfigParser yamlParser &＃61; new YamlConfigParser();Map yamlMap &＃61; yamlParser.doParse(content);doRefresher(yamlMap);}​protected void doRefresher(Map properties) {if (MapUtils.isEmpty(properties)) {log.warn("DynamicTp refresh, empty properties.");return;}PropertiesBinder.bindDtpProperties(properties,dtpProperties);doRefresher(dtpProperties);}​protected void doRefresher(DynamicThreadPoolProperties dtpProperties) {​dynamicThreadPoolRegistry.refresher(dtpProperties);}}

DynamicThreadPoolProperties类在项目启动之初就与.yml文件中的配置项进行映射了&＃xff0c;目前这里是原始值。DynamicThreadPoolRegistry类就是核心类了&＃xff0c;后面会有介绍&＃xff0c;我们先看看refresher方法。

在refresher方法中有个yml文件解析类&＃xff1a;YamlConfigParser类&＃xff1a;

public class YamlConfigParser {public Map doParse(String content) {​if (StringUtils.isEmpty(content)) {return Collections.emptyMap();}YamlPropertiesFactoryBean bean &＃61; new YamlPropertiesFactoryBean();bean.setResources(new ByteArrayResource(content.getBytes()));return bean.getObject();}}

YamlPropertiesFactoryBean是Spring提供给我们使用yml文件解析类&＃xff0c;将yml配置content内容传进去会解析成Map结构&＃xff1a;

这是不是也是个技术点&＃xff01;

紧接着下面还有一个技术点&＃xff0c;我们看下doRefresher方法中&＃xff1a;PropertiesBinder.bindDtpProperties(properties,dtpProperties);

properties&＃xff1a;是从变更的yml文件中解析出来的Map&＃xff0c;是变更之后的新内容。

dtpProperties:是项目启动之初从原始yml文件中解析出来的内容&＃xff0c;是旧内容。

当这两个参数传入到PropertiesBinder.bindDtpProperties(properties,dtpProperties)方法中&＃xff0c;dtpProperties对像中的值就变成了最新的。

这是springBoot给我们提供的&＃xff0c;是不是又学到了。这也是学习开源项目的好处&＃xff0c;会学到一些捷径。

业务开发中如果你需要对比两个对象的属性值差别&＃xff0c;也有一个开源的工具&＃xff1a;equator 感兴趣可以去看下&＃xff0c;这个工具也是在dynamic-tp开源项目中学到的。

下面就到了我们的核心类&＃xff1a;DynamicThreadPoolRegistry

&＃64;Slf4j&＃64;Componentpublic class DynamicThreadPoolRegistry {//这里就不用多说了&＃xff0c;项目中所有配置的线程池都会注入到这个Map结构中&＃xff0c;以线程池的名称为key&＃64;Resourceprivate Map threadPoolExecutorMap &＃61; new HashMap<>();​public void refresher(DynamicThreadPoolProperties dtpProperties) {if (Objects.isNull(dtpProperties) || CollectionUtils.isEmpty(dtpProperties.getExecutors())) {log.warn("DynamicTp refresh, empty threadPoolProperties.");return;}//遍历Map,给线程池重新设置值。dtpProperties.getExecutors().forEach(x -> {if (StringUtils.isBlank(x.getThreadPoolName())) {log.warn("DynamicTp refresh, threadPoolName must not be empty.");return;}ThreadPoolExecutor executor &＃61; threadPoolExecutorMap.get(x.getThreadPoolName());//具体的设置方法refresher(executor, x);});}​private void refresher(ThreadPoolExecutor executor, ThreadPoolProperties properties) {if (properties.getCorePoolSize() <0 ||properties.getMaximumPoolSize() <&＃61; 0 ||properties.getMaximumPoolSize() ) blockingQueue).setCapacity(properties.getQueueCapacity());}}}

原理就是线程池ThreadPoolExecutor自己提供的方法&＃xff1a;

至此动态修改线程池的参数就讲述完了。示例中涉及到代码没有完全贴出来&＃xff0c;此示例代码在github上dynamic-thread-pool

在这里你会看到一个有点陌生的队列VariableLinkedBlockingQueue,这个队列是可以改变容量的&＃xff0c;我们一般创建线程池时使用的队列是LinkedBlockingQueue&＃xff0c;它的默认大小是Integer.MAX_VALUE容易内存溢出&＃xff0c;当然你也可以在构造LinkedBlockingQueue时指定capacity容量大小&＃xff0c;但是capacity是final修饰是没有办法改变的&＃xff0c;VariableLinkedBlockingQueue队列就给它完善了这一点。使用VariableLinkedBlockingQueue队列我们就可以根据自身业务发展情况动态设置队列容量的大小。当然在dynamic-tp项目中还有设计比较巧妙的队列MemorySafeLinkedBlockingQueue&＃xff0c;内存安全队列。这些队列都是可以拿到自己项目中直接使用的。这两个队列在很多开源项目中都使用到了&＃xff0c;设计确实很巧妙。在这个项目中学习到了很多。

到这里我们还有一个Spring提供的扩展技术点没有讲&＃xff1a;

项目启动之初获取Nacos配置&＃xff0c;将yml文件中配置的线程池注册到IOC容器中

我的示例中没有涉及到这一点&＃xff0c;但是dynamic-tp项目中是涉及到了的。下面我们就简单介绍一下这个扩展点&＃xff0c;不光Mybatis使用到这个扩展点&＃xff0c;很多需要融合Spring的开源框架都会使用到&＃xff1a;Nacos&＃xff0c;Dubbo&＃xff0c;Apollo&＃xff0c;RocketMQ太多了就不列举了。

项目启动之初获取Nacos配置这个上面也介绍了一种方式使用&＃64;ConfigurationProperties(prefix &＃61; "spring.dynamic.tp")将配置映射到指定的对象中&＃xff0c;当然这个是有条件的&＃xff0c;就是在使用此对象的类必须是Spring Bean(已经在IOC容器中)。

那如果不是呢&＃xff1f;当然有别的方法&＃xff0c;可以在Environment中获取。既然说到这里就把方法贴出来吧&＃xff1a;

&＃64;Slf4jpublic class DtpBeanDefinitionRegistrar implements EnvironmentAware {​private Environment environment;​&＃64;Overridepublic void setEnvironment(Environment environment) {this.environment &＃61; environment;}​public void registerBeanDefinitions(AnnotationMetadata importingClassMetadata, BeanDefinitionRegistry registry) {​DynamicThreadPoolProperties dtpProperties &＃61; new DynamicThreadPoolProperties();PropertiesBinder.bindDtpProperties(environment, dtpProperties);List executors &＃61; dtpProperties.getExecutors();}}

利用了SpringBoot给我们提供的捷径PropertiesBinder.bindDtpProperties();&＃xff0c;这样dtpProperties对象中就有值了。和上面那个方法只是入参不一样&＃xff0c;上面的入参是一个Map结构&＃xff0c;这里是一个Environment对象。

下面我们继续讲我们的Sping扩展点&＃xff1a;将yml文件中配置的线程池注册到IOC容器中。或者说将外部配置的对象&＃xff0c;再或者说指定的类&＃xff0c;注入到IOC容器中&＃xff0c;在不适用&＃64;Bean&＃xff0c;&＃64;Component&＃xff0c;&＃64;ComponentScan时&＃xff0c;我们使用 ImportBeanDefinitionRegistrar接口。我们先来一个简单的列子&＃xff1a;

首先我定一个注解&＃xff08;模仿Mybatis的&＃64;Mapper&＃xff09;&＃xff1a;

&＃64;Target({ElementType.TYPE})&＃64;Retention(RetentionPolicy.RUNTIME)&＃64;Documented&＃64;Indexedpublic &＃64;interface HXY {​String value() default "";}

public class MyImportBeanDefinitionRegistrar implements ImportBeanDefinitionRegistrar {​&＃64;Overridepublic void registerBeanDefinitions(AnnotationMetadata annotationMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {//Map annotationAttributes &＃61; annotationMetadata.getAnnotationAttributes(ComponentScan.class.getName());//String[] basePackages &＃61; (String[]) annotationAttributes.get("basePackages");​MyClassPathBeanDefinitionScanner scanner &＃61; new MyClassPathBeanDefinitionScanner(beanDefinitionRegistry, false);//指定加上&＃64;HXY注解的类放入IOC容器成为Beanscanner.addIncludeFilter(new AnnotationTypeFilter(HXY.class));scanner.doScan("cn.haoxiaoyong.example");//scanner.doScan(basePackages);}}

这里要借助ClassPathBeanDefinitionScanner

public class MyClassPathBeanDefinitionScanner extends ClassPathBeanDefinitionScanner {public MyClassPathBeanDefinitionScanner(BeanDefinitionRegistry registry, boolean useDefaultFilters) {super(registry, useDefaultFilters);}&＃64;Overrideprotected Set doScan(String... basePackages) {return super.doScan(basePackages);}&＃64;Overridepublic void addIncludeFilter(TypeFilter includeFilter) {super.addIncludeFilter(includeFilter);}}

只要加上&＃64;HXY注解的类就会注册到IOC成为Spring Bean。这也是&＃64;Mapper的实现方式。

如果你不想使用注解可以更简单一点&＃xff1a;

public class MyImportBeanDefinitionRegistrar implements ImportBeanDefinitionRegistrar {&＃64;Overridepublic void registerBeanDefinitions(AnnotationMetadata annotationMetadata, BeanDefinitionRegistry beanDefinitionRegistry) {​beanDefinitionRegistry.registerBeanDefinition("testHxyBean",new RootBeanDefinition(TestHxyBean.class)); }}

注意&＃xff1a;要在启动类&＃xff0c;或者配置类上加&＃64;Import(MyImportBeanDefinitionRegistrar.class)

在dynamic-tp项目中使用了构造方法注入的&＃xff0c;原理就是&＃xff1a;拿到yml文件中配置的线程池构造参数&＃xff0c;我们一般构造一个线程池都需要&＃xff1a;

public ThreadPoolExecutor(int corePoolSize,int maximumPoolSize,long keepAliveTime,TimeUnit unit,BlockingQueue workQueue)

在yml文件中我们也配置了这些构造参数。我们还配置线程池的名称&＃xff0c;然后利用BeanDefinitionBuilder填充ConstructorArgumentValues对象和MutablePropertyValues&＃xff1b;最后还是要调用BeanDefinitionRegistry类的registerBeanDefinition方法注册Bean。具体的可以看下这个开源项目。至此这个扩展点也介绍完了。欢迎mark&＃xff01;