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

优化消息通知模块. (#138)

Browse Source
pull/140/head
chen.ma 2 years ago
parent 3bec271ea1
commit ee60b65102
11 changed files with 79 additions and 45 deletions
Show Stats Download Patch File Download Diff File

12
hippo4j-common/src/main/java/cn/hippo4j/common/notify/BaseSendMessageServiceImpl.java → hippo4j-common/src/main/java/cn/hippo4j/common/notify/HippoBaseSendMessageService.java
Unescape View File

@ -22,7 +22,7 @@ import java.util.Map;
*/
@Slf4j
@RequiredArgsConstructor
public class BaseSendMessageServiceImpl implements HippoSendMessageService, CommandLineRunner {
public class HippoBaseSendMessageService implements HippoSendMessageService, CommandLineRunner {
private final NotifyConfigBuilder notifyConfigBuilder;
@ -38,7 +38,6 @@ public class BaseSendMessageServiceImpl implements HippoSendMessageService, Comm
String buildKey = StrUtil.builder(threadPoolId, "+", "ALARM").toString();
List<NotifyConfigDTO> notifyList = notifyConfigs.get(buildKey);
if (CollUtil.isEmpty(notifyList)) {
log.warn("Please configure alarm notification on the server. key :: [{}]", threadPoolId);
return;
}
@ -113,4 +112,13 @@ public class BaseSendMessageServiceImpl implements HippoSendMessageService, Comm
notifyConfigs.putAll(buildNotify);
}
/**
* Put platform.
*
* @param notifyConfigs
*/
public synchronized void putPlatform(Map<String, List<NotifyConfigDTO>> notifyConfigs) {
this.notifyConfigs.putAll(notifyConfigs);
}
}

74
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/DingAlarmConstants.java
Unescape View File

@ -26,14 +26,14 @@ public class DingAlarmConstants {
/**
* Trace 信息
*/
public static final String DING_ALARM_TIMOUT_TRACE_REPLACE_TXT = "<font color='#708090' size=2>链路信息：%s</font> \n\n";
public static final String DING_ALARM_TIMOUT_TRACE_REPLACE_TXT = "<font color=#708090 size=2>链路信息：%s</font> \n\n";
/**
* 替换任务超时模板
*/
public static final String DING_ALARM_TIMOUT_REPLACE_TXT =
"<font color='#708090' size=2>任务执行时间：%d / ms </font> \n\n" +
"<font color='#708090' size=2>超时时间：%d / ms</font> \n\n" +
"<font color=#708090 size=2>任务执行时间：%d / ms </font> \n\n" +
"<font color=#708090 size=2>超时时间：%d / ms</font> \n\n" +
DING_ALARM_TIMOUT_TRACE_REPLACE_TXT +
" --- \n\n ";
@ -41,30 +41,30 @@ public class DingAlarmConstants {
* 线程池报警通知文本
*/
public static final String DING_ALARM_TXT =
"<font color='#FF0000'>[警报] </font>%s - 动态线程池运行告警（%s） \n\n" +
"**<font color=#FF0000>[警报] </font>%s - 动态线程池运行告警（%s）** \n\n" +
" --- \n\n " +
"<font color='#708090' size=2>线程池ID：%s</font> \n\n " +
"<font color='#708090' size=2>应用名称：%s</font> \n\n " +
"<font color='#778899' size=2>应用实例：%s</font> \n\n " +
"<font color='#778899' size=2>实例标识：%s</font> \n\n " +
"<font color=#708090 size=2>线程池ID：%s</font> \n\n " +
"<font color=#708090 size=2>应用名称：%s</font> \n\n " +
"<font color=#778899 size=2>应用实例：%s</font> \n\n " +
"<font color=#778899 size=2>实例标识：%s</font> \n\n " +
" --- \n\n " +
"<font color='#708090' size=2>核心线程数：%d</font> \n\n " +
"<font color='#708090' size=2>最大线程数：%d</font> \n\n " +
"<font color='#708090' size=2>当前线程数：%d</font> \n\n " +
"<font color='#708090' size=2>活跃线程数：%d</font> \n\n " +
"<font color='#708090' size=2>最大任务数：%d</font> \n\n " +
"<font color='#708090' size=2>线程池任务总量：%d</font> \n\n " +
"<font color=#708090 size=2>核心线程数：%d</font> \n\n " +
"<font color=#708090 size=2>最大线程数：%d</font> \n\n " +
"<font color=#708090 size=2>当前线程数：%d</font> \n\n " +
"<font color=#708090 size=2>活跃线程数：%d</font> \n\n " +
"<font color=#708090 size=2>最大任务数：%d</font> \n\n " +
"<font color=#708090 size=2>线程池任务总量：%d</font> \n\n " +
" --- \n\n " +
"<font color='#708090' size=2>队列类型：%s</font> \n\n " +
"<font color='#708090' size=2>队列容量：%d</font> \n\n " +
"<font color='#708090' size=2>队列元素个数：%d</font> \n\n " +
"<font color='#708090' size=2>队列剩余个数：%d</font> \n\n " +
"<font color=#708090 size=2>队列类型：%s</font> \n\n " +
"<font color=#708090 size=2>队列容量：%d</font> \n\n " +
"<font color=#708090 size=2>队列元素个数：%d</font> \n\n " +
"<font color=#708090 size=2>队列剩余个数：%d</font> \n\n " +
" --- \n\n " +
DING_ALARM_TIMOUT_REPLACE_TXT +
"<font color='#708090' size=2>拒绝策略：%s</font> \n\n" +
"<font color='#708090' size=2>拒绝策略执行次数：</font><font color='#FF0000' size=2>%d</font> \n\n " +
"<font color='#708090' size=2>OWNER：@%s</font> \n\n" +
"<font color='#708090' size=2>提示：%d 分钟内此线程池不会重复告警（可配置）</font> \n\n" +
"<font color=#708090 size=2>拒绝策略：%s</font> \n\n" +
"<font color=#708090 size=2>拒绝策略执行次数：</font><font color=#FF0000 size=2>%d</font> \n\n " +
"<font color=#708090 size=2>OWNER：@%s</font> \n\n" +
"<font color=#708090 size=2>提示：%d 分钟内此线程池不会重复告警（可配置）</font> \n\n" +
" --- \n\n " +
"**播报时间：%s**";
@ -72,25 +72,25 @@ public class DingAlarmConstants {
* 线程池参数变更通知文本
*/
public static final String DING_NOTICE_TXT =
"<font color='#2a9d8f'>[通知] </font>%s - 动态线程池参数变更 \n\n" +
"**<font color=#2a9d8f>[通知] </font>%s - 动态线程池参数变更** \n\n" +
" --- \n\n " +
"<font color='#708090' size=2>线程池ID：%s</font> \n\n " +
"<font color='#708090' size=2>应用名称：%s</font> \n\n " +
"<font color='#778899' size=2>应用实例：%s</font> \n\n " +
"<font color=#708090 size=2>线程池ID：%s</font> \n\n " +
"<font color=#708090 size=2>应用名称：%s</font> \n\n " +
"<font color=#778899 size=2>应用实例：%s</font> \n\n " +
" --- \n\n " +
"<font color='#708090' size=2>核心线程数：%s</font> \n\n " +
"<font color='#708090' size=2>最大线程数：%s</font> \n\n " +
"<font color='#708090' size=2>核心线程超时：%s</font> \n\n " +
"<font color='#708090' size=2>线程存活时间：%s / SECONDS</font> \n\n" +
"<font color=#708090 size=2>核心线程数：%s</font> \n\n " +
"<font color=#708090 size=2>最大线程数：%s</font> \n\n " +
"<font color=#708090 size=2>核心线程超时：%s</font> \n\n " +
"<font color=#708090 size=2>线程存活时间：%s</font> \n\n" +
"<font color=#708090 size=2>执行超时时间：%s</font> \n\n" +
" --- \n\n " +
"<font color='#708090' size=2>队列类型：%s</font> \n\n " +
"<font color='#708090' size=2>队列容量：%s</font> \n\n " +
"<font color=#708090 size=2>队列类型：%s</font> \n\n " +
"<font color=#708090 size=2>队列容量：%s</font> \n\n " +
"<font color=#708090 size=2>AGO 拒绝策略：%s</font> \n\n" +
"<font color=#708090 size=2>NOW 拒绝策略：%s</font> \n\n" +
" --- \n\n " +
"<font color='#708090' size=2>AGO 拒绝策略：%s</font> \n\n" +
"<font color='#708090' size=2>NOW 拒绝策略：%s</font> \n\n" +
" --- \n\n " +
"<font color='#708090' size=2>提示：动态线程池配置变更实时通知（无限制）</font> \n\n" +
"<font color='#708090' size=2>OWNER：@%s</font> \n\n" +
"<font color=#708090 size=2>提示：动态线程池配置变更实时通知（无限制）</font> \n\n" +
"<font color=#708090 size=2>OWNER：@%s</font> \n\n" +
" --- \n\n " +
"**播报时间：%s**";

2
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/DingSendMessageHandler.java
Unescape View File

@ -133,6 +133,8 @@ public class DingSendMessageHandler implements SendMessageHandler<AlarmNotifyReq
changeParameterNotifyRequest.getBeforeAllowsCoreThreadTimeOut() + " ➲ " + changeParameterNotifyRequest.getNowAllowsCoreThreadTimeOut(),
// 线程存活时间
changeParameterNotifyRequest.getBeforeKeepAliveTime() + " ➲ " + changeParameterNotifyRequest.getNowKeepAliveTime(),
// 执行超时时间
changeParameterNotifyRequest.getBeforeExecuteTimeOut() + " ➲ " + changeParameterNotifyRequest.getNowExecuteTimeOut(),
// 阻塞队列
changeParameterNotifyRequest.getBlockingQueueName(),
// 阻塞队列容量

2
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/LarkAlarmConstants.java
Unescape View File

@ -51,6 +51,6 @@ public class LarkAlarmConstants {
/**
* lark notice json str
*/
public static final String LARK_NOTICE_JSON_STR = "{\"msg_type\":\"interactive\",\"card\":{\"config\":{\"wide_screen_mode\":true},\"header\":{\"template\":\"greed\",\"title\":{\"content\":\"[通知] %s 动态线程池参数变更\",\"tag\":\"plain_text\"}},\"elements\":[{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 线程池ID：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 应用名称：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 应用实例：** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 核心线程数：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 最大线程数：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 核心线程超时：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 线程存活时间：** %s / SECONDS\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 队列类型：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 队列容量：** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** AGO 拒绝策略：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** NOW 拒绝策略执行次数：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** OWNER：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 播报时间： ** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"tag\":\"note\",\"elements\":[{\"tag\":\"plain_text\",\"content\":\"提示：动态线程池配置变更实时通知（无限制）\"}]}]}}";
public static final String LARK_NOTICE_JSON_STR = "{\"msg_type\":\"interactive\",\"card\":{\"config\":{\"wide_screen_mode\":true},\"header\":{\"template\":\"greed\",\"title\":{\"content\":\"[通知] %s 动态线程池参数变更\",\"tag\":\"plain_text\"}},\"elements\":[{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 线程池ID：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 应用名称：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 应用实例：** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 核心线程数：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 最大线程数：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 核心线程超时：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 线程存活时间：** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** 队列类型：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 队列容量：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 执行超时时间：** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"fields\":[{\"is_short\":true,\"text\":{\"content\":\"** AGO 拒绝策略：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** NOW 拒绝策略：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** OWNER：** %s\",\"tag\":\"lark_md\"}},{\"is_short\":true,\"text\":{\"content\":\"** 播报时间： ** %s\",\"tag\":\"lark_md\"}}],\"tag\":\"div\"},{\"tag\":\"hr\"},{\"tag\":\"note\",\"elements\":[{\"tag\":\"plain_text\",\"content\":\"提示：动态线程池配置变更实时通知（无限制）\"}]}]}}";
}

2
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/LarkSendMessageHandler.java
Unescape View File

@ -134,6 +134,8 @@ public class LarkSendMessageHandler implements SendMessageHandler<AlarmNotifyReq
changeParameterNotifyRequest.getBlockingQueueName(),
// 阻塞队列容量
changeParameterNotifyRequest.getBeforeQueueCapacity() + " ➲ " + changeParameterNotifyRequest.getNowQueueCapacity(),
// 执行超时时间
changeParameterNotifyRequest.getBeforeExecuteTimeOut() + " ➲ " + changeParameterNotifyRequest.getNowExecuteTimeOut(),
// 拒绝策略
changeParameterNotifyRequest.getBeforeRejectedName(),
changeParameterNotifyRequest.getNowRejectedName(),

5
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/WeChatAlarmConstants.java
Unescape View File

@ -45,7 +45,7 @@ public class WeChatAlarmConstants {
"> 队列元素个数：%s \n" +
"> 队列剩余个数：%s \n" +
"> 拒绝策略：%s \n" +
"> 拒绝策略执行次数：%s \n" +
"> 拒绝策略执行次数：<font color='#FF0000'>%s</font> \n" +
WE_CHAT_ALARM_TIMOUT_REPLACE_TXT +
"> OWNER：<@%s> \n" +
"> 提示：%d 分钟内此线程池不会重复告警（可配置） \n\n" +
@ -62,7 +62,8 @@ public class WeChatAlarmConstants {
"> 核心线程数：%s \n" +
"> 最大线程数：%s \n" +
"> 核心线程超时：%s \n" +
"> 线程存活时间：%s / SECONDS \n" +
"> 线程存活时间：%s \n" +
"> 执行超时时间：%s \n" +
"> 队列类型：%s \n" +
"> 队列容量：%s \n" +
"> AGO 拒绝策略：%s \n" +

2
hippo4j-common/src/main/java/cn/hippo4j/common/notify/platform/WeChatSendMessageHandler.java
Unescape View File

@ -124,6 +124,8 @@ public class WeChatSendMessageHandler implements SendMessageHandler<AlarmNotifyR
changeParameterNotifyRequest.getBeforeAllowsCoreThreadTimeOut() + " ➲ " + changeParameterNotifyRequest.getNowAllowsCoreThreadTimeOut(),
// 线程存活时间
changeParameterNotifyRequest.getBeforeKeepAliveTime() + " ➲ " + changeParameterNotifyRequest.getNowKeepAliveTime(),
// 执行超时时间
changeParameterNotifyRequest.getBeforeExecuteTimeOut() + " ➲ " + changeParameterNotifyRequest.getNowExecuteTimeOut(),
// 阻塞队列
changeParameterNotifyRequest.getBlockingQueueName(),
// 阻塞队列容量

4
hippo4j-common/src/main/java/cn/hippo4j/common/notify/request/ChangeParameterNotifyRequest.java
Unescape View File

@ -34,6 +34,10 @@ public class ChangeParameterNotifyRequest extends BaseNotifyRequest {
private Long nowKeepAliveTime;
private Long beforeExecuteTimeOut;
private Long nowExecuteTimeOut;
private String blockingQueueName;
private Integer beforeQueueCapacity;

10
hippo4j-core/src/main/java/cn/hippo4j/core/executor/ThreadPoolNotifyAlarmHandler.java
Unescape View File

@ -133,6 +133,11 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
* @param threadPoolId
*/
public void checkPoolRejectedAlarm(String threadPoolId) {
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(threadPoolId);
if (!threadPoolNotifyAlarm.getIsAlarm()) {
return;
}
ThreadPoolExecutor threadPoolExecutor = GlobalThreadPoolManage.getExecutorService(threadPoolId).getExecutor();
checkPoolRejectedAlarm(threadPoolId, threadPoolExecutor);
}
@ -160,6 +165,11 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
* @param threadPoolExecutor
*/
public void asyncSendExecuteTimeOutAlarm(String threadPoolId, long executeTime, long executeTimeOut, ThreadPoolExecutor threadPoolExecutor) {
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(threadPoolId);
if (!threadPoolNotifyAlarm.getIsAlarm()) {
return;
}
if (threadPoolExecutor instanceof DynamicThreadPoolExecutor) {
try {
AlarmNotifyRequest alarmNotifyRequest = buildAlarmNotifyReq(threadPoolExecutor);

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

@ -36,7 +36,7 @@ public class MessageNotifyConfiguration {
@Bean
public HippoSendMessageService hippoSendMessageService(NotifyConfigBuilder notifyConfigBuilder,
AlarmControlHandler alarmControlHandler) {
return new BaseSendMessageServiceImpl(notifyConfigBuilder, alarmControlHandler);
return new HippoBaseSendMessageService(notifyConfigBuilder, alarmControlHandler);
}
@Bean

9
hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/starter/core/ServerThreadPoolDynamicRefresh.java
Unescape View File

@ -60,13 +60,14 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
long originalKeepAliveTime = executor.getKeepAliveTime(TimeUnit.SECONDS);
boolean originalAllowCoreThreadTimeOut = executor.allowsCoreThreadTimeOut();
String originalRejected;
Long originalExecuteTimeOut = null;
RejectedExecutionHandler rejectedExecutionHandler = executor.getRejectedExecutionHandler();
if (executor instanceof AbstractDynamicExecutorSupport) {
DynamicThreadPoolExecutor dynamicExecutor = (DynamicThreadPoolExecutor) executor;
rejectedExecutionHandler = dynamicExecutor.getRedundancyHandler();
originalExecuteTimeOut = dynamicExecutor.getExecuteTimeOut();
}
originalRejected = rejectedExecutionHandler.getClass().getSimpleName();
String originalRejected = rejectedExecutionHandler.getClass().getSimpleName();
// Send change message.
ChangeParameterNotifyRequest request = new ChangeParameterNotifyRequest();
@ -77,6 +78,7 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
request.setBlockingQueueName(originalQuery);
request.setBeforeQueueCapacity(originalCapacity);
request.setBeforeRejectedName(originalRejected);
request.setBeforeExecuteTimeOut(originalExecuteTimeOut);
request.setThreadPoolId(threadPoolId);
changePoolInfo(executor, parameter);
@ -88,6 +90,7 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
request.setNowKeepAliveTime(afterExecutor.getKeepAliveTime(TimeUnit.SECONDS));
request.setNowQueueCapacity((afterExecutor.getQueue().remainingCapacity() + afterExecutor.getQueue().size()));
request.setNowRejectedName(RejectedTypeEnum.getRejectedNameByType(parameter.getRejectedType()));
request.setNowExecuteTimeOut(originalExecuteTimeOut);
threadPoolNotifyAlarmHandler.sendPoolConfigChange(request);
log.info(
@ -97,6 +100,7 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
"\n queueType :: [{}]" +
"\n capacity :: [{}]" +
"\n keepAliveTime :: [{}]" +
"\n executeTimeOut :: [{}]" +
"\n rejectedType :: [{}]" +
"\n allowCoreThreadTimeOut :: [{}]",
threadPoolId.toUpperCase(),
@ -106,6 +110,7 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
String.format("%s => %s", originalCapacity,
(afterExecutor.getQueue().remainingCapacity() + afterExecutor.getQueue().size())),
String.format("%s => %s", originalKeepAliveTime, afterExecutor.getKeepAliveTime(TimeUnit.SECONDS)),
String.format("%s => %s", originalExecuteTimeOut, originalExecuteTimeOut),
String.format("%s => %s", originalRejected, RejectedTypeEnum.getRejectedNameByType(parameter.getRejectedType())),
String.format("%s => %s", originalAllowCoreThreadTimeOut, EnableEnum.getBool(parameter.getAllowCoreThreadTimeOut()))
);

Write Preview
Loading…
Cancel
Save