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

refactor: refactor DynamicThreadPoolRefreshListener function code

Browse Source
pull/1555/head
Song 2 weeks ago
parent a5ec18d7fc
commit ba7538de8e
1 changed files with 109 additions and 66 deletions
Show Stats Download Patch File Download Diff File

175
starters/threadpool/config/src/main/java/cn/hippo4j/config/springboot/starter/refresher/event/DynamicThreadPoolRefreshListener.java
Unescape View File

@ -38,7 +38,6 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.core.annotation.Order; import org.springframework.core.annotation.Order;
import java.util.ArrayList;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
@ -46,6 +45,7 @@ import java.util.Optional;
import java.util.concurrent.RejectedExecutionHandler; import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER; import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_THREAD_POOL_TEXT; import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_THREAD_POOL_TEXT;
@ -164,48 +164,63 @@ public class DynamicThreadPoolRefreshListener extends AbstractRefreshListener<Ex
* @param executorProperties * @param executorProperties
*/ */
private void checkNotifyConsistencyAndReplace(ExecutorProperties executorProperties) { private void checkNotifyConsistencyAndReplace(ExecutorProperties executorProperties) {
boolean checkNotifyConfig = false;
boolean checkNotifyAlarm = false;
Map<String, List<NotifyConfigDTO>> newDynamicThreadPoolNotifyMap = Map<String, List<NotifyConfigDTO>> newDynamicThreadPoolNotifyMap =
configModeNotifyConfigBuilder.buildSingleNotifyConfig(executorProperties); configModeNotifyConfigBuilder.buildSingleNotifyConfig(executorProperties);
Map<String, List<NotifyConfigDTO>> notifyConfigs = threadPoolBaseSendMessageService.getNotifyConfigs(); Map<String, List<NotifyConfigDTO>> notifyConfigs = threadPoolBaseSendMessageService.getNotifyConfigs();
if (CollectionUtil.isNotEmpty(notifyConfigs)) {
for (Map.Entry<String, List<NotifyConfigDTO>> each : newDynamicThreadPoolNotifyMap.entrySet()) { boolean checkNotifyConfig = checkAndReplaceNotifyConfig(newDynamicThreadPoolNotifyMap, notifyConfigs);
if (checkNotifyConfig) { boolean checkNotifyAlarm = checkAndReplaceNotifyAlarm(executorProperties);
break;
} if (checkNotifyConfig || checkNotifyAlarm) {
List<NotifyConfigDTO> notifyConfigDTOS = notifyConfigs.get(each.getKey()); log.info("[{}] Dynamic thread pool notification property changes.", executorProperties.getThreadPoolId());
for (NotifyConfigDTO notifyConfig : each.getValue()) {
if (!notifyConfigDTOS.contains(notifyConfig)) {
checkNotifyConfig = true;
break;
}
}
}
} }
if (checkNotifyConfig) { }
configModeNotifyConfigBuilder.initCacheAndLock(newDynamicThreadPoolNotifyMap);
threadPoolBaseSendMessageService.putPlatform(newDynamicThreadPoolNotifyMap); private boolean checkAndReplaceNotifyConfig(Map<String, List<NotifyConfigDTO>> newConfigs,
Map<String, List<NotifyConfigDTO>> currentConfigs) {
if (CollectionUtil.isEmpty(currentConfigs)) {
return false;
} }
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(executorProperties.getThreadPoolId());
if (threadPoolNotifyAlarm != null) { for (Map.Entry<String, List<NotifyConfigDTO>> entry : newConfigs.entrySet()) {
Boolean isAlarm = executorProperties.getAlarm(); String key = entry.getKey();
Integer activeAlarm = executorProperties.getActiveAlarm(); List<NotifyConfigDTO> newNotifyConfigList = entry.getValue();
Integer capacityAlarm = executorProperties.getCapacityAlarm(); List<NotifyConfigDTO> currentNotifyConfigList = currentConfigs.get(key);
if ((isAlarm != null && !Objects.equals(isAlarm, threadPoolNotifyAlarm.getAlarm()))
|| (activeAlarm != null && !Objects.equals(activeAlarm, threadPoolNotifyAlarm.getActiveAlarm())) if (currentNotifyConfigList == null || !currentNotifyConfigList.containsAll(newNotifyConfigList)) {
|| (capacityAlarm != null && !Objects.equals(capacityAlarm, threadPoolNotifyAlarm.getCapacityAlarm()))) { configModeNotifyConfigBuilder.initCacheAndLock(newConfigs);
checkNotifyAlarm = true; threadPoolBaseSendMessageService.putPlatform(newConfigs);
threadPoolNotifyAlarm.setAlarm(Optional.ofNullable(isAlarm).orElse(threadPoolNotifyAlarm.getAlarm())); return true;
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.", executorProperties.getThreadPoolId()); return false;
}
private boolean checkAndReplaceNotifyAlarm(ExecutorProperties executorProperties) {
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(executorProperties.getThreadPoolId());
if (threadPoolNotifyAlarm == null) {
return false;
}
boolean checkNotifyAlarm = false;
Boolean isAlarm = executorProperties.getAlarm();
Integer activeAlarm = executorProperties.getActiveAlarm();
Integer capacityAlarm = executorProperties.getCapacityAlarm();
if ((isAlarm != null && !Objects.equals(isAlarm, threadPoolNotifyAlarm.getAlarm()))
|| (activeAlarm != null && !Objects.equals(activeAlarm, threadPoolNotifyAlarm.getActiveAlarm()))
|| (capacityAlarm != null && !Objects.equals(capacityAlarm, threadPoolNotifyAlarm.getCapacityAlarm()))) {
checkNotifyAlarm = true;
threadPoolNotifyAlarm.setAlarm(isAlarm != null ? isAlarm : threadPoolNotifyAlarm.getAlarm());
threadPoolNotifyAlarm.setActiveAlarm(activeAlarm != null ? activeAlarm : threadPoolNotifyAlarm.getActiveAlarm());
threadPoolNotifyAlarm.setCapacityAlarm(capacityAlarm != null ? capacityAlarm : threadPoolNotifyAlarm.getCapacityAlarm());
} }
return checkNotifyAlarm;
} }
/** /**
* Check consistency. * Check consistency.
* *
@ -216,19 +231,26 @@ public class DynamicThreadPoolRefreshListener extends AbstractRefreshListener<Ex
ThreadPoolExecutorHolder executorHolder = ThreadPoolExecutorRegistry.getHolder(threadPoolId); ThreadPoolExecutorHolder executorHolder = ThreadPoolExecutorRegistry.getHolder(threadPoolId);
ExecutorProperties beforeProperties = executorHolder.getExecutorProperties(); ExecutorProperties beforeProperties = executorHolder.getExecutorProperties();
ThreadPoolExecutor executor = executorHolder.getExecutor(); ThreadPoolExecutor executor = executorHolder.getExecutor();
if (executor == null) { if (executor == null) {
return false; return false;
} }
boolean result = (properties.getCorePoolSize() != null && !Objects.equals(beforeProperties.getCorePoolSize(), properties.getCorePoolSize()))
|| (properties.getMaximumPoolSize() != null && !Objects.equals(beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize())) return Stream.of(
|| (properties.getAllowCoreThreadTimeOut() != null && !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())) hasPropertyChanged(beforeProperties.getCorePoolSize(), properties.getCorePoolSize()),
|| (properties.getExecuteTimeOut() != null && !Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut())) hasPropertyChanged(beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize()),
|| (properties.getKeepAliveTime() != null && !Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())) hasPropertyChanged(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut()),
|| (properties.getRejectedHandler() != null && !Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())) hasPropertyChanged(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()),
|| hasPropertyChanged(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime()),
((properties.getQueueCapacity() != null && !Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity()) hasPropertyChanged(beforeProperties.getRejectedHandler(), properties.getRejectedHandler()),
&& Objects.equals(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.getName(), executor.getQueue().getClass().getSimpleName()))); isQueueCapacityChanged(beforeProperties, properties, executor)
return result; ).anyMatch(Boolean::booleanValue);
}
private boolean isQueueCapacityChanged(ExecutorProperties beforeProperties, ExecutorProperties properties, ThreadPoolExecutor executor) {
return properties.getQueueCapacity() != null &&
!Objects.equals(beforeProperties.getQueueCapacity(), properties.getQueueCapacity()) &&
Objects.equals(BlockingQueueTypeEnum.RESIZABLE_LINKED_BLOCKING_QUEUE.getName(), executor.getQueue().getClass().getSimpleName());
} }
/** /**
@ -238,41 +260,62 @@ public class DynamicThreadPoolRefreshListener extends AbstractRefreshListener<Ex
* @param properties * @param properties
*/ */
private void dynamicRefreshPool(String threadPoolId, ExecutorProperties properties) { private void dynamicRefreshPool(String threadPoolId, ExecutorProperties properties) {
ExecutorProperties beforeProperties = ThreadPoolExecutorRegistry.getHolder(threadPoolId).getExecutorProperties(); ThreadPoolExecutorHolder holder = ThreadPoolExecutorRegistry.getHolder(threadPoolId);
ThreadPoolExecutor executor = ThreadPoolExecutorRegistry.getHolder(threadPoolId).getExecutor(); ExecutorProperties beforeProperties = holder.getExecutorProperties();
if (properties.getMaximumPoolSize() != null && properties.getCorePoolSize() != null) { ThreadPoolExecutor executor = holder.getExecutor();
ThreadPoolExecutorUtil.safeSetPoolSize(executor, properties.getCorePoolSize(), properties.getMaximumPoolSize());
if (executor == null) {
log.warn("Executor is null for threadPoolId: {}", threadPoolId);
return;
}
setPoolSizes(executor, properties);
updateExecutorProperties(executor, beforeProperties, properties);
updateQueueCapacity(executor, beforeProperties, properties);
}
private void setPoolSizes(ThreadPoolExecutor executor, ExecutorProperties properties) {
Integer corePoolSize = properties.getCorePoolSize();
Integer maximumPoolSize = properties.getMaximumPoolSize();
if (corePoolSize != null && maximumPoolSize != null) {
ThreadPoolExecutorUtil.safeSetPoolSize(executor, corePoolSize, maximumPoolSize);
} else { } else {
if (properties.getMaximumPoolSize() != null) { if (maximumPoolSize != null) {
executor.setMaximumPoolSize(properties.getMaximumPoolSize()); executor.setMaximumPoolSize(maximumPoolSize);
} }
if (properties.getCorePoolSize() != null) { if (corePoolSize != null) {
executor.setCorePoolSize(properties.getCorePoolSize()); executor.setCorePoolSize(corePoolSize);
} }
} }
if (properties.getAllowCoreThreadTimeOut() != null && !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())) { }
private void updateExecutorProperties(ThreadPoolExecutor executor, ExecutorProperties beforeProperties, ExecutorProperties properties) {
if (hasPropertyChanged(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())) {
executor.allowCoreThreadTimeOut(properties.getAllowCoreThreadTimeOut()); executor.allowCoreThreadTimeOut(properties.getAllowCoreThreadTimeOut());
} }
if (properties.getExecuteTimeOut() != null && !Objects.equals(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut())) { if (hasPropertyChanged(beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()) && executor instanceof DynamicThreadPoolExecutor) {
if (executor instanceof DynamicThreadPoolExecutor) { ((DynamicThreadPoolExecutor) executor).setExecuteTimeOut(properties.getExecuteTimeOut());
((DynamicThreadPoolExecutor) executor).setExecuteTimeOut(properties.getExecuteTimeOut());
}
} }
if (properties.getRejectedHandler() != null && !Objects.equals(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())) { if (hasPropertyChanged(beforeProperties.getRejectedHandler(), properties.getRejectedHandler())) {
RejectedExecutionHandler rejectedExecutionHandler = RejectedPolicyTypeEnum.createPolicy(properties.getRejectedHandler()); RejectedExecutionHandler rejectedExecutionHandler = RejectedPolicyTypeEnum.createPolicy(properties.getRejectedHandler());
executor.setRejectedExecutionHandler(rejectedExecutionHandler); executor.setRejectedExecutionHandler(rejectedExecutionHandler);
} }
if (properties.getKeepAliveTime() != null && !Objects.equals(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())) { if (hasPropertyChanged(beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime())) {
executor.setKeepAliveTime(properties.getKeepAliveTime(), TimeUnit.SECONDS); executor.setKeepAliveTime(properties.getKeepAliveTime(), TimeUnit.SECONDS);
} }
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) { private void updateQueueCapacity(ThreadPoolExecutor executor, ExecutorProperties beforeProperties, ExecutorProperties properties) {
ResizableCapacityLinkedBlockingQueue<?> queue = (ResizableCapacityLinkedBlockingQueue<?>) executor.getQueue(); if (hasPropertyChanged(beforeProperties.getQueueCapacity(), properties.getQueueCapacity())
queue.setCapacity(properties.getQueueCapacity()); && executor.getQueue() instanceof ResizableCapacityLinkedBlockingQueue) {
} else { ((ResizableCapacityLinkedBlockingQueue<?>) executor.getQueue()).setCapacity(properties.getQueueCapacity());
log.warn("The queue length cannot be modified. Queue type mismatch. Current queue type: {}", executor.getQueue().getClass().getSimpleName()); } else {
} log.warn("The queue length cannot be modified. Queue type mismatch. Current queue type: {}", executor.getQueue().getClass().getSimpleName());
} }
} }
private <T> boolean hasPropertyChanged(T before, T after) {
return after != null && !Objects.equals(before, after);
}
} }

Write Preview
Loading…
Cancel
Save