msb-public
/
hippo4j
mirror of https://github.com/longtai-cn/hippo4j
1
0
Code Issues Projects Releases Wiki Activity

When defining a dynamic thread pool, abstract the default configuration (#572)

Browse Source
pull/602/head
chen.ma 2 years ago
parent 29a47565b2
commit 7459495757
5 changed files with 179 additions and 169 deletions
Show Stats Download Patch File Download Diff File

28
hippo4j-spring-boot/hippo4j-config-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/config/BootstrapConfigProperties.java
Unescape View File

@ -107,43 +107,23 @@ public class BootstrapConfigProperties implements BootstrapPropertiesInterface {
*/
private List<NotifyPlatformProperties> notifyPlatforms;
/**
* Whether to enable thread pool running alarm.
*/
private Boolean alarm = Boolean.TRUE;
/**
* Check thread pool running status interval.
*/
private Integer checkStateInterval;
/**
* Active alarm.
*/
private Integer activeAlarm;
/**
* Capacity alarm.
*/
private Integer capacityAlarm;
/**
* Thread pool run alarm interval. unit: s
*/
private Integer alarmInterval;
/**
* Receives.
* Default dynamic thread pool configuration.
*/
private String receives;
private ExecutorProperties defaultExecutor;
/**
* Executors.
* Dynamic thread pool configuration collection.
*/
private List<ExecutorProperties> executors;
/**
* Adapter executors
* Tripartite framework thread pool adaptation set.
*/
private List<AdapterExecutorProperties> adapterExecutors;
}

2
hippo4j-spring-boot/hippo4j-config-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/config/DynamicThreadPoolNotifyProperties.java
Unescape View File

@ -30,7 +30,7 @@ import lombok.NoArgsConstructor;
public class DynamicThreadPoolNotifyProperties {
/**
* Interval
* Thread pool run alarm interval. unit: s
*/
private Integer interval;

41
hippo4j-spring-boot/hippo4j-config-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/notify/CoreNotifyConfigBuilder.java
Unescape View File

@ -45,13 +45,13 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
private final AlarmControlHandler alarmControlHandler;
private final BootstrapConfigProperties bootstrapConfigProperties;
private final BootstrapConfigProperties configProperties;
@Override
public Map<String, List<NotifyConfigDTO>> buildNotify() {
Map<String, List<NotifyConfigDTO>> resultMap = Maps.newHashMap();
boolean globalAlarm = bootstrapConfigProperties.getAlarm();
List<ExecutorProperties> executors = bootstrapConfigProperties.getExecutors();
boolean globalAlarm = Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getAlarm()).orElse(true);
List<ExecutorProperties> executors = configProperties.getExecutors();
if (CollectionUtil.isEmpty(executors)) {
log.warn("Failed to build notify, executors configuration is empty.");
return resultMap;
@ -60,8 +60,8 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
if (!globalAlarm && CollectionUtil.isEmpty(actual)) {
return resultMap;
}
for (ExecutorProperties executor : executors) {
Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig = buildSingleNotifyConfig(executor);
for (ExecutorProperties executorProperties : executors) {
Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig = buildSingleNotifyConfig(executorProperties);
initCacheAndLock(buildSingleNotifyConfig);
resultMap.putAll(buildSingleNotifyConfig);
}
@ -71,15 +71,15 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
/**
* Build single notify config.
*
* @param executor
* @param executorProperties
* @return
*/
public Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig(ExecutorProperties executor) {
public Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig(ExecutorProperties executorProperties) {
Map<String, List<NotifyConfigDTO>> resultMap = Maps.newHashMap();
String threadPoolId = executor.getThreadPoolId();
String threadPoolId = executorProperties.getThreadPoolId();
String alarmBuildKey = threadPoolId + "+ALARM";
List<NotifyConfigDTO> alarmNotifyConfigs = Lists.newArrayList();
List<NotifyPlatformProperties> notifyPlatforms = bootstrapConfigProperties.getNotifyPlatforms();
List<NotifyPlatformProperties> notifyPlatforms = configProperties.getNotifyPlatforms();
for (NotifyPlatformProperties platformProperties : notifyPlatforms) {
NotifyConfigDTO notifyConfig = new NotifyConfigDTO();
notifyConfig.setPlatform(platformProperties.getPlatform());
@ -87,11 +87,11 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
notifyConfig.setType("ALARM");
notifyConfig.setSecret(platformProperties.getSecret());
notifyConfig.setSecretKey(getToken(platformProperties));
int interval = Optional.ofNullable(executor.getNotify())
int interval = Optional.ofNullable(executorProperties.getNotify())
.map(each -> each.getInterval())
.orElseGet(() -> bootstrapConfigProperties.getAlarmInterval() != null ? bootstrapConfigProperties.getAlarmInterval() : 5);
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getNotify()).map(each -> each.getInterval()).orElse(5));
notifyConfig.setInterval(interval);
notifyConfig.setReceives(buildReceive(executor, platformProperties));
notifyConfig.setReceives(buildReceive(executorProperties));
alarmNotifyConfigs.add(notifyConfig);
}
resultMap.put(alarmBuildKey, alarmNotifyConfigs);
@ -104,7 +104,7 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
notifyConfig.setType("CONFIG");
notifyConfig.setSecretKey(getToken(platformProperties));
notifyConfig.setSecret(platformProperties.getSecret());
notifyConfig.setReceives(buildReceive(executor, platformProperties));
notifyConfig.setReceives(buildReceive(executorProperties));
changeNotifyConfigs.add(notifyConfig);
}
resultMap.put(changeBuildKey, changeNotifyConfigs);
@ -118,17 +118,12 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
.forEach(each -> alarmControlHandler.initCacheAndLock(each.getTpId(), each.getPlatform(), each.getInterval())));
}
private String buildReceive(ExecutorProperties executor, NotifyPlatformProperties platformProperties) {
String receive;
if (executor.getNotify() != null) {
receive = executor.getNotify().getReceives();
if (StrUtil.isBlank(receive)) {
receive = bootstrapConfigProperties.getReceives();
}
} else {
receive = bootstrapConfigProperties.getReceives();
private String buildReceive(ExecutorProperties executorProperties) {
String receives = Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getNotify()).map(each -> each.getReceives()).orElse("");
if (executorProperties.getNotify() != null && StringUtil.isNotEmpty(executorProperties.getNotify().getReceives())) {
receives = executorProperties.getNotify().getReceives();
}
return receive;
return receives;
}
private String getToken(NotifyPlatformProperties platformProperties) {

123
hippo4j-spring-boot/hippo4j-config-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/refresher/event/DynamicThreadPoolRefreshListener.java
Unescape View File

@ -17,15 +17,15 @@
package cn.hippo4j.core.springboot.starter.refresher.event;
import cn.hippo4j.common.executor.support.BlockingQueueTypeEnum;
import cn.hippo4j.common.executor.support.RejectedPolicyTypeEnum;
import cn.hippo4j.common.executor.support.ResizableCapacityLinkedBlockingQueue;
import cn.hippo4j.common.toolkit.CollectionUtil;
import cn.hippo4j.core.executor.DynamicThreadPoolExecutor;
import cn.hippo4j.core.executor.ThreadPoolNotifyAlarmHandler;
import cn.hippo4j.core.executor.manage.GlobalNotifyAlarmManage;
import cn.hippo4j.core.executor.manage.GlobalThreadPoolManage;
import cn.hippo4j.core.executor.support.AbstractDynamicExecutorSupport;
import cn.hippo4j.common.executor.support.BlockingQueueTypeEnum;
import cn.hippo4j.common.executor.support.RejectedPolicyTypeEnum;
import cn.hippo4j.common.executor.support.ResizableCapacityLinkedBlockingQueue;
import cn.hippo4j.core.proxy.RejectedProxyUtil;
import cn.hippo4j.core.springboot.starter.config.BootstrapConfigProperties;
import cn.hippo4j.core.springboot.starter.config.ExecutorProperties;
@ -44,6 +44,7 @@ import org.springframework.core.annotation.Order;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
@ -68,40 +69,52 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
private final Hippo4jBaseSendMessageService hippo4jBaseSendMessageService;
@Override
public void onApplicationEvent(Hippo4jConfigDynamicRefreshEvent threadPoolDynamicRefreshEvent) {
BootstrapConfigProperties bindableConfigProperties = threadPoolDynamicRefreshEvent.getBootstrapConfigProperties();
public void onApplicationEvent(Hippo4jConfigDynamicRefreshEvent event) {
BootstrapConfigProperties bindableConfigProperties = event.getBootstrapConfigProperties();
List<ExecutorProperties> executors = bindableConfigProperties.getExecutors();
for (ExecutorProperties properties : executors) {
String threadPoolId = properties.getThreadPoolId();
// Check whether the notification configuration is consistent.
// this operation will not trigger the notification.
/**
* Check whether the notification configuration is consistent, this operation will not trigger the notification.
*/
checkNotifyConsistencyAndReplace(properties);
if (!checkConsistency(threadPoolId, properties)) {
continue;
}
// refresh executor pool.
dynamicRefreshPool(threadPoolId, properties);
// old properties.
ExecutorProperties beforeProperties = GlobalCoreThreadPoolManage.getProperties(properties.getThreadPoolId());
// refresh executor properties.
GlobalCoreThreadPoolManage.refresh(threadPoolId, properties);
GlobalCoreThreadPoolManage.refresh(threadPoolId, failDefaultExecutorProperties(beforeProperties, properties));
ChangeParameterNotifyRequest changeRequest = buildChangeRequest(beforeProperties, properties);
log.info(CHANGE_THREAD_POOL_TEXT,
threadPoolId,
String.format(CHANGE_DELIMITER, beforeProperties.getCorePoolSize(), properties.getCorePoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getQueueCapacity(), properties.getQueueCapacity()),
String.format(CHANGE_DELIMITER, beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime()),
String.format(CHANGE_DELIMITER, beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()),
String.format(CHANGE_DELIMITER, beforeProperties.getRejectedHandler(), properties.getRejectedHandler()),
String.format(CHANGE_DELIMITER, beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut()));
String.format(CHANGE_DELIMITER, beforeProperties.getCorePoolSize(), changeRequest.getNowCorePoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getMaximumPoolSize(), changeRequest.getNowMaximumPoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getQueueCapacity(), changeRequest.getNowQueueCapacity()),
String.format(CHANGE_DELIMITER, beforeProperties.getKeepAliveTime(), changeRequest.getNowKeepAliveTime()),
String.format(CHANGE_DELIMITER, beforeProperties.getExecuteTimeOut(), changeRequest.getNowExecuteTimeOut()),
String.format(CHANGE_DELIMITER, beforeProperties.getRejectedHandler(), changeRequest.getNowRejectedName()),
String.format(CHANGE_DELIMITER, beforeProperties.getAllowCoreThreadTimeOut(), changeRequest.getNowAllowsCoreThreadTimeOut()));
try {
threadPoolNotifyAlarmHandler.sendPoolConfigChange(newChangeRequest(beforeProperties, properties));
threadPoolNotifyAlarmHandler.sendPoolConfigChange(changeRequest);
} catch (Throwable ex) {
log.error("Failed to send Chang smart application listener notice. Message: {}", ex.getMessage());
}
}
}
private ExecutorProperties failDefaultExecutorProperties(ExecutorProperties beforeProperties, ExecutorProperties properties) {
return ExecutorProperties.builder()
.corePoolSize(Optional.ofNullable(properties.getCorePoolSize()).orElse(beforeProperties.getCorePoolSize()))
.maximumPoolSize(Optional.ofNullable(properties.getMaximumPoolSize()).orElse(beforeProperties.getMaximumPoolSize()))
.queueCapacity(Optional.ofNullable(properties.getQueueCapacity()).orElse(beforeProperties.getQueueCapacity()))
.keepAliveTime(Optional.ofNullable(properties.getKeepAliveTime()).orElse(beforeProperties.getKeepAliveTime()))
.executeTimeOut(Optional.ofNullable(properties.getExecuteTimeOut()).orElse(beforeProperties.getExecuteTimeOut()))
.rejectedHandler(Optional.ofNullable(properties.getRejectedHandler()).orElse(beforeProperties.getRejectedHandler()))
.allowCoreThreadTimeOut(Optional.ofNullable(properties.getAllowCoreThreadTimeOut()).orElse(beforeProperties.getAllowCoreThreadTimeOut()))
.threadPoolId(beforeProperties.getThreadPoolId())
.build();
}
/**
* Construct change parameter notify request instance.
*
@ -109,23 +122,23 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
* @param properties new properties
* @return instance
*/
private ChangeParameterNotifyRequest newChangeRequest(ExecutorProperties beforeProperties, ExecutorProperties properties) {
private ChangeParameterNotifyRequest buildChangeRequest(ExecutorProperties beforeProperties, ExecutorProperties properties) {
ChangeParameterNotifyRequest changeParameterNotifyRequest = ChangeParameterNotifyRequest.builder()
.blockingQueueName(beforeProperties.getBlockingQueue())
.beforeCorePoolSize(beforeProperties.getCorePoolSize())
.beforeMaximumPoolSize(beforeProperties.getMaximumPoolSize())
.beforeAllowsCoreThreadTimeOut(beforeProperties.getAllowCoreThreadTimeOut())
.beforeKeepAliveTime(beforeProperties.getKeepAliveTime())
.blockingQueueName(beforeProperties.getBlockingQueue())
.beforeQueueCapacity(beforeProperties.getQueueCapacity())
.beforeRejectedName(beforeProperties.getRejectedHandler())
.beforeExecuteTimeOut(beforeProperties.getExecuteTimeOut())
.nowCorePoolSize(properties.getCorePoolSize())
.nowMaximumPoolSize(properties.getMaximumPoolSize())
.nowAllowsCoreThreadTimeOut(properties.getAllowCoreThreadTimeOut())
.nowKeepAliveTime(properties.getKeepAliveTime())
.nowQueueCapacity(properties.getQueueCapacity())
.nowRejectedName(properties.getRejectedHandler())
.nowExecuteTimeOut(properties.getExecuteTimeOut())
.nowCorePoolSize(Optional.ofNullable(properties.getCorePoolSize()).orElse(beforeProperties.getCorePoolSize()))
.nowMaximumPoolSize(Optional.ofNullable(properties.getMaximumPoolSize()).orElse(beforeProperties.getMaximumPoolSize()))
.nowAllowsCoreThreadTimeOut(Optional.ofNullable(properties.getAllowCoreThreadTimeOut()).orElse(beforeProperties.getAllowCoreThreadTimeOut()))
.nowKeepAliveTime(Optional.ofNullable(properties.getKeepAliveTime()).orElse(beforeProperties.getKeepAliveTime()))
.nowQueueCapacity(Optional.ofNullable(properties.getQueueCapacity()).orElse(beforeProperties.getQueueCapacity()))
.nowRejectedName(Optional.ofNullable(properties.getRejectedHandler()).orElse(beforeProperties.getRejectedHandler()))
.nowExecuteTimeOut(Optional.ofNullable(properties.getExecuteTimeOut()).orElse(beforeProperties.getExecuteTimeOut()))
.build();
changeParameterNotifyRequest.setThreadPoolId(beforeProperties.getThreadPoolId());
return changeParameterNotifyRequest;
@ -134,13 +147,13 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
/**
* Check notify consistency and replace.
*
* @param properties
* @param executorProperties
*/
private void checkNotifyConsistencyAndReplace(ExecutorProperties properties) {
private void checkNotifyConsistencyAndReplace(ExecutorProperties executorProperties) {
boolean checkNotifyConfig = false;
boolean checkNotifyAlarm = false;
List<String> changeKeys = Lists.newArrayList();
Map<String, List<NotifyConfigDTO>> newDynamicThreadPoolNotifyMap = coreNotifyConfigBuilder.buildSingleNotifyConfig(properties);
Map<String, List<NotifyConfigDTO>> newDynamicThreadPoolNotifyMap = coreNotifyConfigBuilder.buildSingleNotifyConfig(executorProperties);
Map<String, List<NotifyConfigDTO>> notifyConfigs = hippo4jBaseSendMessageService.getNotifyConfigs();
if (CollectionUtil.isNotEmpty(notifyConfigs)) {
for (Map.Entry<String, List<NotifyConfigDTO>> each : newDynamicThreadPoolNotifyMap.entrySet()) {
@ -161,22 +174,22 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
coreNotifyConfigBuilder.initCacheAndLock(newDynamicThreadPoolNotifyMap);
hippo4jBaseSendMessageService.putPlatform(newDynamicThreadPoolNotifyMap);
}
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(properties.getThreadPoolId());
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(executorProperties.getThreadPoolId());
if (threadPoolNotifyAlarm != null) {
boolean isAlarm = properties.getAlarm();
Integer activeAlarm = properties.getActiveAlarm();
Integer capacityAlarm = properties.getCapacityAlarm();
if (threadPoolNotifyAlarm.getAlarm() != isAlarm
|| threadPoolNotifyAlarm.getActiveAlarm() != activeAlarm
|| threadPoolNotifyAlarm.getCapacityAlarm() != capacityAlarm) {
Boolean isAlarm = executorProperties.getAlarm();
Integer activeAlarm = executorProperties.getActiveAlarm();
Integer capacityAlarm = executorProperties.getCapacityAlarm();
if ((isAlarm != null && isAlarm != threadPoolNotifyAlarm.getAlarm())
|| (activeAlarm != null && activeAlarm != threadPoolNotifyAlarm.getActiveAlarm())
|| (capacityAlarm != null && capacityAlarm != threadPoolNotifyAlarm.getCapacityAlarm())) {
checkNotifyAlarm = true;
threadPoolNotifyAlarm.setAlarm(isAlarm);
threadPoolNotifyAlarm.setActiveAlarm(activeAlarm);
threadPoolNotifyAlarm.setCapacityAlarm(capacityAlarm);
threadPoolNotifyAlarm.setAlarm(Optional.ofNullable(isAlarm).orElse(threadPoolNotifyAlarm.getAlarm()));
threadPoolNotifyAlarm.setActiveAlarm(Optional.ofNullable(activeAlarm).orElse(threadPoolNotifyAlarm.getActiveAlarm()));
threadPoolNotifyAlarm.setCapacityAlarm(Optional.ofNullable(capacityAlarm).orElse(threadPoolNotifyAlarm.getCapacityAlarm()));
}
}
if (checkNotifyConfig || checkNotifyAlarm) {
log.info("[{}] Dynamic thread pool notification property changes.", properties.getThreadPoolId());
log.info("[{}] Dynamic thread pool notification property changes.", executorProperties.getThreadPoolId());
}
}
@ -192,15 +205,15 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
if (executor == null) {
return false;
}
boolean result = !Objects.equals(beforeProperties.getCorePoolSize(), properties.getCorePoolSize())
|| !Objects.equals(beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize())
|| !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())
|| !Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut())
|| !Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())
|| !Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())
boolean result = (properties.getCorePoolSize() != null && !Objects.equals(beforeProperties.getCorePoolSize(), properties.getCorePoolSize()))
|| (properties.getMaximumPoolSize() != null && !Objects.equals(beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize()))
|| (properties.getAllowCoreThreadTimeOut() != null && !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut()))
|| (properties.getExecuteTimeOut() != null && !Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()))
|| (properties.getKeepAliveTime() != null && !Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime()))
|| (properties.getRejectedHandler() != null && !Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler()))
||
(!Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity())
&& Objects.equals(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.getName(), executor.getQueue().getClass().getSimpleName()));
((properties.getQueueCapacity() != null && !Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity())
&& Objects.equals(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.getName(), executor.getQueue().getClass().getSimpleName())));
return result;
}
@ -229,15 +242,15 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
executor.setCorePoolSize(properties.getCorePoolSize());
}
}
if (!Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())) {
if (properties.getAllowCoreThreadTimeOut() != null && !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())) {
executor.allowCoreThreadTimeOut(properties.getAllowCoreThreadTimeOut());
}
if (!Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut())) {
if (properties.getExecuteTimeOut() != null && !Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut())) {
if (executor instanceof AbstractDynamicExecutorSupport) {
((DynamicThreadPoolExecutor) executor).setExecuteTimeOut(properties.getExecuteTimeOut());
}
}
if (!Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())) {
if (properties.getRejectedHandler() != null && !Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())) {
RejectedExecutionHandler rejectedExecutionHandler = RejectedPolicyTypeEnum.createPolicy(properties.getRejectedHandler());
if (executor instanceof AbstractDynamicExecutorSupport) {
DynamicThreadPoolExecutor dynamicExecutor = (DynamicThreadPoolExecutor) executor;
@ -247,10 +260,10 @@ public class DynamicThreadPoolRefreshListener implements ApplicationListener<Hip
}
executor.setRejectedExecutionHandler(rejectedExecutionHandler);
}
if (!Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())) {
if (properties.getKeepAliveTime() != null && !Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())) {
executor.setKeepAliveTime(properties.getKeepAliveTime(), TimeUnit.SECONDS);
}
if (!Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity())
if (properties.getQueueCapacity() != null && !Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity())
&& Objects.equals(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.getName(), executor.getQueue().getClass().getSimpleName())) {
if (executor.getQueue() instanceof ResizableCapacityLinkedBlockingQueue) {
ResizableCapacityLinkedBlockingQueue<?> queue = (ResizableCapacityLinkedBlockingQueue<?>) executor.getQueue();

154
hippo4j-spring-boot/hippo4j-config-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/support/DynamicThreadPoolPostProcessor.java
Unescape View File

@ -26,7 +26,8 @@ import cn.hippo4j.core.executor.DynamicThreadPoolExecutor;
import cn.hippo4j.core.executor.DynamicThreadPoolWrapper;
import cn.hippo4j.core.executor.manage.GlobalNotifyAlarmManage;
import cn.hippo4j.core.executor.manage.GlobalThreadPoolManage;
import cn.hippo4j.core.executor.support.*;
import cn.hippo4j.core.executor.support.CommonDynamicThreadPool;
import cn.hippo4j.core.executor.support.ThreadPoolBuilder;
import cn.hippo4j.core.executor.support.adpter.DynamicThreadPoolAdapterChoose;
import cn.hippo4j.core.springboot.starter.config.BootstrapConfigProperties;
import cn.hippo4j.core.springboot.starter.config.DynamicThreadPoolNotifyProperties;
@ -52,7 +53,7 @@ import java.util.concurrent.TimeUnit;
@AllArgsConstructor
public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {
private final BootstrapConfigProperties bootstrapConfigProperties;
private final BootstrapConfigProperties configProperties;
@Override
public Object postProcessBeforeInitialization(Object bean, String beanName) {
@ -87,20 +88,11 @@ public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {
}
if (bean instanceof DynamicThreadPoolWrapper) {
DynamicThreadPoolWrapper wrap = (DynamicThreadPoolWrapper) bean;
registerAndSubscribe(wrap);
fillPoolAndRegister(wrap);
}
return bean;
}
/**
* Register and subscribe.
*
* @param dynamicThreadPoolWrap
*/
protected void registerAndSubscribe(DynamicThreadPoolWrapper dynamicThreadPoolWrap) {
fillPoolAndRegister(dynamicThreadPoolWrap);
}
/**
* Fill the thread pool and register.
*
@ -110,57 +102,30 @@ public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {
String threadPoolId = dynamicThreadPoolWrapper.getThreadPoolId();
ThreadPoolExecutor newDynamicPoolExecutor = dynamicThreadPoolWrapper.getExecutor();
ExecutorProperties executorProperties = null;
if (null != bootstrapConfigProperties.getExecutors()) {
executorProperties = bootstrapConfigProperties.getExecutors()
if (configProperties.getExecutors() != null) {
executorProperties = configProperties.getExecutors()
.stream()
.filter(each -> Objects.equals(threadPoolId, each.getThreadPoolId()))
.findFirst()
.orElse(null);
if (executorProperties != null) {
try {
BlockingQueue workQueue = BlockingQueueTypeEnum.createBlockingQueue(executorProperties.getBlockingQueue(), executorProperties.getQueueCapacity());
String threadNamePrefix = executorProperties.getThreadNamePrefix();
newDynamicPoolExecutor = ThreadPoolBuilder.builder()
.dynamicPool()
.workQueue(workQueue)
.threadFactory(StringUtil.isNotBlank(threadNamePrefix) ? threadNamePrefix : threadPoolId)
.executeTimeOut(Optional.ofNullable(executorProperties.getExecuteTimeOut()).orElse(0L))
.poolThreadSize(executorProperties.getCorePoolSize(), executorProperties.getMaximumPoolSize())
.keepAliveTime(executorProperties.getKeepAliveTime(), TimeUnit.SECONDS)
.rejected(RejectedPolicyTypeEnum.createPolicy(executorProperties.getRejectedHandler()))
.allowCoreThreadTimeOut(executorProperties.getAllowCoreThreadTimeOut())
.build();
} catch (Exception ex) {
log.error("Failed to initialize thread pool configuration. error: {}", ex);
} finally {
if (Objects.isNull(dynamicThreadPoolWrapper.getExecutor())) {
dynamicThreadPoolWrapper.setExecutor(CommonDynamicThreadPool.getInstance(threadPoolId));
}
dynamicThreadPoolWrapper.setInitFlag(Boolean.TRUE);
.orElseThrow(() -> new RuntimeException("The thread pool id does not exist in the configuration."));
try {
newDynamicPoolExecutor = buildNewDynamicThreadPool(executorProperties);
} catch (Exception ex) {
log.error("Failed to initialize thread pool configuration. error: {}", ex);
} finally {
if (Objects.isNull(dynamicThreadPoolWrapper.getExecutor())) {
dynamicThreadPoolWrapper.setExecutor(CommonDynamicThreadPool.getInstance(threadPoolId));
}
dynamicThreadPoolWrapper.setInitFlag(Boolean.TRUE);
}
if (dynamicThreadPoolWrapper.getExecutor() instanceof AbstractDynamicExecutorSupport) {
DynamicThreadPoolNotifyProperties notify = Optional.ofNullable(executorProperties).map(ExecutorProperties::getNotify).orElse(null);
boolean isAlarm = Optional.ofNullable(executorProperties.getAlarm())
.orElseGet(() -> bootstrapConfigProperties.getAlarm() != null ? bootstrapConfigProperties.getAlarm() : true);
int activeAlarm = Optional.ofNullable(executorProperties.getActiveAlarm())
.orElseGet(() -> bootstrapConfigProperties.getActiveAlarm() != null ? bootstrapConfigProperties.getActiveAlarm() : 80);
int capacityAlarm = Optional.ofNullable(executorProperties.getCapacityAlarm())
.orElseGet(() -> bootstrapConfigProperties.getCapacityAlarm() != null ? bootstrapConfigProperties.getCapacityAlarm() : 80);
int interval = Optional.ofNullable(notify)
.map(each -> each.getInterval()).orElseGet(() -> bootstrapConfigProperties.getAlarmInterval() != null ? bootstrapConfigProperties.getAlarmInterval() : 5);
String receive = Optional.ofNullable(notify)
.map(each -> each.getReceives()).orElseGet(() -> StringUtil.isNotEmpty(bootstrapConfigProperties.getReceives()) ? bootstrapConfigProperties.getReceives() : "");
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = new ThreadPoolNotifyAlarm(isAlarm, activeAlarm, capacityAlarm);
threadPoolNotifyAlarm.setInterval(interval);
threadPoolNotifyAlarm.setReceives(receive);
GlobalNotifyAlarmManage.put(threadPoolId, threadPoolNotifyAlarm);
TaskDecorator taskDecorator = ((DynamicThreadPoolExecutor) dynamicThreadPoolWrapper.getExecutor()).getTaskDecorator();
((DynamicThreadPoolExecutor) newDynamicPoolExecutor).setTaskDecorator(taskDecorator);
long awaitTerminationMillis = ((DynamicThreadPoolExecutor) dynamicThreadPoolWrapper.getExecutor()).awaitTerminationMillis;
boolean waitForTasksToCompleteOnShutdown = ((DynamicThreadPoolExecutor) dynamicThreadPoolWrapper.getExecutor()).waitForTasksToCompleteOnShutdown;
((DynamicThreadPoolExecutor) newDynamicPoolExecutor).setSupportParam(awaitTerminationMillis, waitForTasksToCompleteOnShutdown);
}
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = buildThreadPoolNotifyAlarm(executorProperties);
GlobalNotifyAlarmManage.put(threadPoolId, threadPoolNotifyAlarm);
DynamicThreadPoolExecutor actualDynamicThreadPoolExecutor = (DynamicThreadPoolExecutor) dynamicThreadPoolWrapper.getExecutor();
TaskDecorator taskDecorator = actualDynamicThreadPoolExecutor.getTaskDecorator();
((DynamicThreadPoolExecutor) newDynamicPoolExecutor).setTaskDecorator(taskDecorator);
long awaitTerminationMillis = actualDynamicThreadPoolExecutor.awaitTerminationMillis;
boolean waitForTasksToCompleteOnShutdown = actualDynamicThreadPoolExecutor.waitForTasksToCompleteOnShutdown;
((DynamicThreadPoolExecutor) newDynamicPoolExecutor).setSupportParam(awaitTerminationMillis, waitForTasksToCompleteOnShutdown);
dynamicThreadPoolWrapper.setExecutor(newDynamicPoolExecutor);
}
GlobalThreadPoolManage.registerPool(dynamicThreadPoolWrapper.getThreadPoolId(), dynamicThreadPoolWrapper);
@ -168,17 +133,14 @@ public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {
threadPoolId,
executorProperties == null
? buildExecutorProperties(threadPoolId, newDynamicPoolExecutor)
: executorProperties);
: buildActualExecutorProperties(executorProperties));
return newDynamicPoolExecutor;
}
/**
* Build executor properties.
*
* @param threadPoolId
* @param executor
* @return
*/
private ExecutorProperties buildActualExecutorProperties(ExecutorProperties executorProperties) {
return Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> buildExecutorProperties(executorProperties)).orElse(executorProperties);
}
private ExecutorProperties buildExecutorProperties(String threadPoolId, ThreadPoolExecutor executor) {
ExecutorProperties executorProperties = new ExecutorProperties();
BlockingQueue<Runnable> blockingQueue = executor.getQueue();
@ -197,4 +159,64 @@ public final class DynamicThreadPoolPostProcessor implements BeanPostProcessor {
.setThreadPoolId(threadPoolId);
return executorProperties;
}
private ThreadPoolExecutor buildNewDynamicThreadPool(ExecutorProperties executorProperties) {
String threadNamePrefix = executorProperties.getThreadNamePrefix();
ExecutorProperties newExecutorProperties = buildExecutorProperties(executorProperties);
ThreadPoolExecutor newDynamicPoolExecutor = ThreadPoolBuilder.builder()
.dynamicPool()
.poolThreadSize(newExecutorProperties.getCorePoolSize(), newExecutorProperties.getMaximumPoolSize())
.workQueue(BlockingQueueTypeEnum.createBlockingQueue(newExecutorProperties.getBlockingQueue(), newExecutorProperties.getQueueCapacity()))
.threadFactory(StringUtil.isNotBlank(threadNamePrefix) ? threadNamePrefix : executorProperties.getThreadPoolId())
.executeTimeOut(newExecutorProperties.getExecuteTimeOut())
.keepAliveTime(newExecutorProperties.getKeepAliveTime(), TimeUnit.SECONDS)
.rejected(RejectedPolicyTypeEnum.createPolicy(newExecutorProperties.getRejectedHandler()))
.allowCoreThreadTimeOut(newExecutorProperties.getAllowCoreThreadTimeOut())
.build();
return newDynamicPoolExecutor;
}
private ExecutorProperties buildExecutorProperties(ExecutorProperties executorProperties) {
ExecutorProperties newExecutorProperties = ExecutorProperties.builder()
.corePoolSize(Optional.ofNullable(executorProperties.getCorePoolSize())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getCorePoolSize()).get()))
.maximumPoolSize(Optional.ofNullable(executorProperties.getMaximumPoolSize())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getMaximumPoolSize()).get()))
.allowCoreThreadTimeOut(Optional.ofNullable(executorProperties.getAllowCoreThreadTimeOut())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getAllowCoreThreadTimeOut()).get()))
.keepAliveTime(Optional.ofNullable(executorProperties.getKeepAliveTime())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getKeepAliveTime()).get()))
.blockingQueue(Optional.ofNullable(executorProperties.getBlockingQueue())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getBlockingQueue()).get()))
.executeTimeOut(Optional.ofNullable(executorProperties.getExecuteTimeOut())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getExecuteTimeOut()).orElse(0L)))
.queueCapacity(Optional.ofNullable(executorProperties.getQueueCapacity())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getQueueCapacity()).get()))
.rejectedHandler(Optional.ofNullable(executorProperties.getRejectedHandler())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getRejectedHandler()).get()))
.threadNamePrefix(StringUtil.isBlank(executorProperties.getThreadNamePrefix()) ? executorProperties.getThreadPoolId() : executorProperties.getThreadNamePrefix())
.threadPoolId(executorProperties.getThreadPoolId())
.build();
return newExecutorProperties;
}
private ThreadPoolNotifyAlarm buildThreadPoolNotifyAlarm(ExecutorProperties executorProperties) {
DynamicThreadPoolNotifyProperties notify = Optional.ofNullable(executorProperties).map(ExecutorProperties::getNotify).orElse(null);
boolean isAlarm = Optional.ofNullable(executorProperties.getAlarm())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getAlarm()).orElse(true));
int activeAlarm = Optional.ofNullable(executorProperties.getActiveAlarm())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getActiveAlarm()).orElse(80));
int capacityAlarm = Optional.ofNullable(executorProperties.getCapacityAlarm())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getCapacityAlarm()).orElse(80));
int interval = Optional.ofNullable(notify)
.map(each -> each.getInterval())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getNotify()).map(each -> each.getInterval()).orElse(5));
String receive = Optional.ofNullable(notify)
.map(each -> each.getReceives())
.orElseGet(() -> Optional.ofNullable(configProperties.getDefaultExecutor()).map(each -> each.getNotify()).map(each -> each.getReceives()).orElse(""));
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = new ThreadPoolNotifyAlarm(isAlarm, activeAlarm, capacityAlarm);
threadPoolNotifyAlarm.setInterval(interval);
threadPoolNotifyAlarm.setReceives(receive);
return threadPoolNotifyAlarm;
}
}

Write Preview
Loading…
Cancel
Save