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

Merge pull request #411 from mabaiwan/develop

Browse Source 
Notify related parameters to add dynamic change function (#407)
pull/413/head
小马哥 2 years ago committed by GitHub
parent e2a7756981 7fbddb1f46
commit 2260365eae
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 107 additions and 31 deletions
Show Stats Download Patch File Download Diff File

22
hippo4j-message/src/main/java/cn/hippo4j/message/dto/NotifyConfigDTO.java
Unescape View File

@ -19,62 +19,64 @@ package cn.hippo4j.message.dto;
import cn.hippo4j.message.enums.NotifyTypeEnum;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.experimental.Accessors;
/**
* Notify config dto.
*/
@Data
@EqualsAndHashCode
@Accessors(chain = true)
public class NotifyConfigDTO {
/**
* 租户id
* Tenant id
*/
private String tenantId;
/**
* 项目id
* Item id
*/
private String itemId;
/**
* 线程池id
* Thread-pool id
*/
private String tpId;
/**
* 通知平台
* Platform
*/
private String platform;
/**
* 通知类型
* Type
*/
private String type;
/**
* 密钥
* Secret key
*/
private String secretKey;
/**
* 加签
* Secret
*/
private String secret;
/**
* 报警间隔
* Interval
*/
private Integer interval;
/**
* 接收者
* Receives
*/
private String receives;
/**
* 报警类型
* Type enum
*/
private NotifyTypeEnum typeEnum;
}

4
hippo4j-message/src/main/java/cn/hippo4j/message/service/HippoBaseSendMessageService.java
Unescape View File

@ -27,6 +27,7 @@ import cn.hippo4j.message.request.ChangeParameterNotifyRequest;
import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.StrUtil;
import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
@ -45,7 +46,8 @@ public class HippoBaseSendMessageService implements HippoSendMessageService, Com
private final AlarmControlHandler alarmControlHandler;
private final Map<String, List<NotifyConfigDTO>> notifyConfigs = Maps.newHashMap();
@Getter
public final Map<String, List<NotifyConfigDTO>> notifyConfigs = Maps.newHashMap();
private final Map<String, SendMessageHandler> sendMessageHandlers = Maps.newHashMap();

17
hippo4j-message/src/main/java/cn/hippo4j/message/service/ThreadPoolNotifyAlarm.java
Unescape View File

@ -33,36 +33,39 @@ import java.util.Map;
public class ThreadPoolNotifyAlarm {
/**
* isAlarm
* Is alarm
*/
@NonNull
private Boolean isAlarm;
/**
* activeAlarm
* Active alarm
*/
@NonNull
private Integer activeAlarm;
/**
* capacityAlarm
* Capacity alarm
*/
@NonNull
private Integer capacityAlarm;
/**
* interval
* Interval
*/
private Integer interval;
/**
* receive
* Receive
*/
private String receive;
/**
* receives
* ps：暂不启用该配置，后续如果开发邮箱时或许有用
* Receives
*
* <p>
* Do not enable this configuration for the time being, it may be useful if you develop mailboxes in the future.
* </p>
*/
@Deprecated
private Map<String, String> receives;

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

@ -36,6 +36,7 @@ import cn.hippo4j.core.springboot.starter.support.ThreadPoolAdapterRegister;
import cn.hippo4j.message.api.NotifyConfigBuilder;
import cn.hippo4j.message.config.MessageConfiguration;
import cn.hippo4j.message.service.AlarmControlHandler;
import cn.hippo4j.message.service.HippoBaseSendMessageService;
import cn.hippo4j.message.service.HippoSendMessageService;
import lombok.AllArgsConstructor;
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
@ -107,8 +108,11 @@ public class DynamicThreadPoolCoreAutoConfiguration {
}
@Bean
public ExecutorsListener hippo4jExecutorsListener(ThreadPoolNotifyAlarmHandler threadPoolNotifyAlarmHandler) {
return new ExecutorsListener(threadPoolNotifyAlarmHandler);
@SuppressWarnings("all")
public ExecutorsListener hippo4jExecutorsListener(ThreadPoolNotifyAlarmHandler threadPoolNotifyAlarmHandler,
CoreNotifyConfigBuilder coreNotifyConfigBuilder,
HippoBaseSendMessageService hippoBaseSendMessageService) {
return new ExecutorsListener(threadPoolNotifyAlarmHandler, coreNotifyConfigBuilder, hippoBaseSendMessageService);
}
@Bean

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

@ -64,7 +64,9 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
return resultMap;
}
for (ExecutorProperties executor : executors) {
resultMap.putAll(buildSingleNotifyConfig(executor));
Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig = buildSingleNotifyConfig(executor);
initCacheAndLock(buildSingleNotifyConfig);
resultMap.putAll(buildSingleNotifyConfig);
}
return resultMap;
}
@ -80,7 +82,6 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
String threadPoolId = executor.getThreadPoolId();
String alarmBuildKey = threadPoolId + "+ALARM";
List<NotifyConfigDTO> alarmNotifyConfigs = Lists.newArrayList();
List<NotifyPlatformProperties> notifyPlatforms = bootstrapCoreProperties.getNotifyPlatforms();
for (NotifyPlatformProperties platformProperties : notifyPlatforms) {
NotifyConfigDTO notifyConfig = new NotifyConfigDTO();
@ -97,10 +98,8 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
alarmNotifyConfigs.add(notifyConfig);
}
resultMap.put(alarmBuildKey, alarmNotifyConfigs);
String changeBuildKey = threadPoolId + "+CONFIG";
List<NotifyConfigDTO> changeNotifyConfigs = Lists.newArrayList();
for (NotifyPlatformProperties platformProperties : notifyPlatforms) {
NotifyConfigDTO notifyConfig = new NotifyConfigDTO();
notifyConfig.setPlatform(platformProperties.getPlatform());
@ -112,13 +111,14 @@ public class CoreNotifyConfigBuilder implements NotifyConfigBuilder {
changeNotifyConfigs.add(notifyConfig);
}
resultMap.put(changeBuildKey, changeNotifyConfigs);
return resultMap;
}
resultMap.forEach(
public void initCacheAndLock(Map<String, List<NotifyConfigDTO>> buildSingleNotifyConfig) {
buildSingleNotifyConfig.forEach(
(key, val) -> val.stream()
.filter(each -> StrUtil.equals("ALARM", each.getType()))
.forEach(each -> alarmControlHandler.initCacheAndLock(each.getTpId(), each.getPlatform(), each.getInterval())));
return resultMap;
}
private String buildReceive(ExecutorProperties executor, NotifyPlatformProperties platformProperties) {

73
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/springboot/starter/refresher/event/ExecutorsListener.java
Unescape View File

@ -17,9 +17,10 @@
package cn.hippo4j.core.springboot.starter.refresher.event;
import cn.hippo4j.message.request.ChangeParameterNotifyRequest;
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.core.executor.support.QueueTypeEnum;
@ -28,13 +29,20 @@ import cn.hippo4j.core.executor.support.ResizableCapacityLinkedBlockingQueue;
import cn.hippo4j.core.proxy.RejectedProxyUtil;
import cn.hippo4j.core.springboot.starter.config.BootstrapCoreProperties;
import cn.hippo4j.core.springboot.starter.config.ExecutorProperties;
import cn.hippo4j.core.springboot.starter.notify.CoreNotifyConfigBuilder;
import cn.hippo4j.core.springboot.starter.support.GlobalCoreThreadPoolManage;
import cn.hippo4j.message.dto.NotifyConfigDTO;
import cn.hippo4j.message.request.ChangeParameterNotifyRequest;
import cn.hippo4j.message.service.HippoBaseSendMessageService;
import cn.hippo4j.message.service.ThreadPoolNotifyAlarm;
import com.google.common.collect.Lists;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationListener;
import org.springframework.core.annotation.Order;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.RejectedExecutionHandler;
import java.util.concurrent.ThreadPoolExecutor;
@ -55,20 +63,27 @@ public class ExecutorsListener implements ApplicationListener<Hippo4jCoreDynamic
private final ThreadPoolNotifyAlarmHandler threadPoolNotifyAlarmHandler;
private final CoreNotifyConfigBuilder coreNotifyConfigBuilder;
private final HippoBaseSendMessageService hippoBaseSendMessageService;
@Override
public void onApplicationEvent(Hippo4jCoreDynamicRefreshEvent threadPoolDynamicRefreshEvent) {
BootstrapCoreProperties bindableCoreProperties = threadPoolDynamicRefreshEvent.getBootstrapCoreProperties();
List<ExecutorProperties> executors = bindableCoreProperties.getExecutors();
for (ExecutorProperties properties : executors) {
String threadPoolId = properties.getThreadPoolId();
// Check whether the notification configuration is consistent.
// this operation will not trigger the notification.
checkNotifyConsistencyAndReplace(properties);
if (!checkConsistency(threadPoolId, properties)) {
continue;
}
// refresh executor pool
// refresh executor pool.
dynamicRefreshPool(threadPoolId, properties);
// old properties
// old properties.
ExecutorProperties beforeProperties = GlobalCoreThreadPoolManage.getProperties(properties.getThreadPoolId());
// refresh executor properties
// refresh executor properties.
GlobalCoreThreadPoolManage.refresh(threadPoolId, properties);
log.info(CHANGE_THREAD_POOL_TEXT,
threadPoolId.toUpperCase(),
@ -116,6 +131,56 @@ public class ExecutorsListener implements ApplicationListener<Hippo4jCoreDynamic
return changeRequest;
}
/**
* Check notify consistency and replace.
*
* @param properties
*/
private void checkNotifyConsistencyAndReplace(ExecutorProperties properties) {
boolean checkNotifyConfig = false;
boolean checkNotifyAlarm = false;
List<String> changeKeys = Lists.newArrayList();
Map<String, List<NotifyConfigDTO>> newDynamicThreadPoolNotifyMap = coreNotifyConfigBuilder.buildSingleNotifyConfig(properties);
Map<String, List<NotifyConfigDTO>> notifyConfigs = hippoBaseSendMessageService.getNotifyConfigs();
if (CollectionUtil.isNotEmpty(notifyConfigs)) {
for (Map.Entry<String, List<NotifyConfigDTO>> each : newDynamicThreadPoolNotifyMap.entrySet()) {
if (checkNotifyConfig) {
break;
}
List<NotifyConfigDTO> notifyConfigDTOS = notifyConfigs.get(each.getKey());
for (NotifyConfigDTO notifyConfig : each.getValue()) {
if (!notifyConfigDTOS.contains(notifyConfig)) {
checkNotifyConfig = true;
changeKeys.add(each.getKey());
break;
}
}
}
}
if (checkNotifyConfig) {
coreNotifyConfigBuilder.initCacheAndLock(newDynamicThreadPoolNotifyMap);
hippoBaseSendMessageService.putPlatform(newDynamicThreadPoolNotifyMap);
}
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(properties.getThreadPoolId());
if (threadPoolNotifyAlarm != null && properties.getNotify() != null) {
ThreadPoolNotifyAlarm notify = properties.getNotify();
boolean isAlarm = notify.getIsAlarm();
Integer activeAlarm = notify.getActiveAlarm();
Integer capacityAlarm = notify.getCapacityAlarm();
if (threadPoolNotifyAlarm.getIsAlarm() != isAlarm
|| threadPoolNotifyAlarm.getActiveAlarm() != activeAlarm
|| threadPoolNotifyAlarm.getCapacityAlarm() != capacityAlarm) {
checkNotifyAlarm = true;
threadPoolNotifyAlarm.setIsAlarm(isAlarm);
threadPoolNotifyAlarm.setActiveAlarm(activeAlarm);
threadPoolNotifyAlarm.setCapacityAlarm(capacityAlarm);
}
}
if (checkNotifyConfig || checkNotifyAlarm) {
log.info("[{}] Dynamic thread pool notification property changes.", properties.getThreadPoolId());
}
}
/**
* Check consistency.
*

Write Preview
Loading…
Cancel
Save