Merge pull request #519 from mabaiwan/develop

hippo4j server dynamic registration thread pool refactoring

hippo4j-common/src/main/java/cn/hippo4j/common/model/register/notify/DynamicThreadPoolRegisterServerNotifyParameter.java

@@ -37,9 +37,9 @@ public class DynamicThreadPoolRegisterServerNotifyParameter {
     private String platform;
 
     /**
-     * Secret key
+     * Access token
      */
-    private String secretKey;
+    private String accessToken;
 
     /**
      * Interval

hippo4j-config/src/main/java/cn/hippo4j/config/service/biz/impl/ConfigServiceImpl.java

@@ -165,7 +165,7 @@ public class ConfigServiceImpl implements ConfigService {
                         .setTpId(configAllInfo.getTpId())
                         .setPlatform(serverNotifyParameter.getPlatform())
                         .setReceives(serverNotifyParameter.getReceives())
-                        .setSecretKey(serverNotifyParameter.getSecretKey());
+                        .setSecretKey(serverNotifyParameter.getAccessToken());
                 if (Objects.equals(each, "ALARM")) {
                     notifyReqDTO.setInterval(serverNotifyParameter.getInterval());
                     notifyReqDTO.setAlarmType(true);

hippo4j-example/hippo4j-example-core/src/main/java/cn/hippo4j/example/core/inittest/RegisterDynamicThreadPoolTest.java

@@ -63,7 +63,7 @@ public class RegisterDynamicThreadPoolTest {
                 .build();
         DynamicThreadPoolRegisterServerNotifyParameter serverNotifyParameter = DynamicThreadPoolRegisterServerNotifyParameter.builder()
                 .platform(NotifyPlatformEnum.WECHAT.name())
-                .secretKey("xxx")
+                .accessToken("7487d0a0-20ec-40ab-b67b-ce68db406b37")
                 .interval(5)
                 .receives("chen.ma")
                 .build();

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/config/DynamicThreadPoolAutoConfiguration.java

@@ -47,6 +47,7 @@ import cn.hippo4j.springboot.starter.notify.ServerNotifyConfigBuilder;
 import cn.hippo4j.springboot.starter.remote.HttpAgent;
 import cn.hippo4j.springboot.starter.remote.HttpScheduledHealthCheck;
 import cn.hippo4j.springboot.starter.remote.ServerHealthCheck;
+import cn.hippo4j.springboot.starter.support.DynamicThreadPoolPostProcessor;
 import lombok.AllArgsConstructor;
 import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/core/DynamicThreadPoolConfigService.java

@@ -21,12 +21,15 @@ import cn.hippo4j.common.model.ThreadPoolParameterInfo;
 import cn.hippo4j.common.model.register.DynamicThreadPoolRegisterParameter;
 import cn.hippo4j.common.model.register.DynamicThreadPoolRegisterWrapper;
 import cn.hippo4j.common.toolkit.Assert;
+import cn.hippo4j.common.toolkit.BooleanUtil;
 import cn.hippo4j.common.toolkit.JSONUtil;
 import cn.hippo4j.common.web.base.Result;
 import cn.hippo4j.common.web.exception.ServiceException;
 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.service.AbstractDynamicThreadPoolService;
+import cn.hippo4j.message.service.ThreadPoolNotifyAlarm;
 import cn.hippo4j.springboot.starter.config.BootstrapProperties;
 import cn.hippo4j.springboot.starter.event.ApplicationCompleteEvent;
 import cn.hippo4j.springboot.starter.remote.HttpAgent;
@@ -55,9 +58,20 @@ public class DynamicThreadPoolConfigService extends AbstractDynamicThreadPoolSer
 
     @Override
     public ThreadPoolExecutor registerDynamicThreadPool(DynamicThreadPoolRegisterWrapper registerWrapper) {
+        ThreadPoolExecutor dynamicThreadPoolExecutor = registerExecutor(registerWrapper);
+        subscribeConfig(registerWrapper);
+        putNotifyAlarmConfig(registerWrapper);
+        return dynamicThreadPoolExecutor;
+    }
+
+    @Override
+    public void onApplicationEvent(ApplicationCompleteEvent event) {
+        clientWorker.notifyApplicationComplete();
+    }
+
+    private ThreadPoolExecutor registerExecutor(DynamicThreadPoolRegisterWrapper registerWrapper) {
         DynamicThreadPoolRegisterParameter registerParameter = registerWrapper.getDynamicThreadPoolRegisterParameter();
         checkThreadPoolParameter(registerParameter);
-        ThreadPoolParameterInfo parameter = JSONUtil.parseObject(JSONUtil.toJSONString(registerParameter), ThreadPoolParameterInfo.class);
         String threadPoolId = registerParameter.getThreadPoolId();
         try {
             failDynamicThreadPoolRegisterWrapper(registerWrapper);
@@ -69,20 +83,27 @@ public class DynamicThreadPoolConfigService extends AbstractDynamicThreadPoolSer
             log.error("Dynamic thread pool registration execution error: {}", threadPoolId, ex);
             throw ex;
         }
+        ThreadPoolParameterInfo parameter = JSONUtil.parseObject(JSONUtil.toJSONString(registerParameter), ThreadPoolParameterInfo.class);
         ThreadPoolExecutor dynamicThreadPoolExecutor = buildDynamicThreadPoolExecutor(registerParameter);
         DynamicThreadPoolWrapper dynamicThreadPoolWrapper = DynamicThreadPoolWrapper.builder()
                 .threadPoolId(threadPoolId)
                 .executor(dynamicThreadPoolExecutor)
                 .build();
         GlobalThreadPoolManage.register(threadPoolId, parameter, dynamicThreadPoolWrapper);
-        dynamicThreadPoolSubscribeConfig.subscribeConfig(threadPoolId);
-        clientWorker.addCacheDataIfAbsent(properties.getNamespace(), properties.getItemId(), threadPoolId);
         return dynamicThreadPoolExecutor;
     }
 
-    @Override
-    public void onApplicationEvent(ApplicationCompleteEvent event) {
-        clientWorker.notifyApplicationComplete();
+    private void subscribeConfig(DynamicThreadPoolRegisterWrapper registerWrapper) {
+        dynamicThreadPoolSubscribeConfig.subscribeConfig(registerWrapper.getDynamicThreadPoolRegisterParameter().getThreadPoolId());
+    }
+
+    private void putNotifyAlarmConfig(DynamicThreadPoolRegisterWrapper registerWrapper) {
+        DynamicThreadPoolRegisterParameter registerParameter = registerWrapper.getDynamicThreadPoolRegisterParameter();
+        ThreadPoolNotifyAlarm threadPoolNotifyAlarm = new ThreadPoolNotifyAlarm(
+                BooleanUtil.toBoolean(String.valueOf(registerParameter.getIsAlarm())),
+                registerParameter.getActiveAlarm(),
+                registerParameter.getCapacityAlarm());
+        GlobalNotifyAlarmManage.put(registerParameter.getThreadPoolId(), threadPoolNotifyAlarm);
     }
 
     private void checkThreadPoolParameter(DynamicThreadPoolRegisterParameter registerParameter) {

hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/springboot/starter/support/DynamicThreadPoolPostProcessor.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package cn.hippo4j.springboot.starter.core;
+package cn.hippo4j.springboot.starter.support;
 
 import cn.hippo4j.common.config.ApplicationContextHolder;
 import cn.hippo4j.common.constant.Constants;
@@ -34,6 +34,7 @@ import cn.hippo4j.core.executor.support.adpter.DynamicThreadPoolAdapterChoose;
 import cn.hippo4j.core.toolkit.inet.DynamicThreadPoolAnnotationUtil;
 import cn.hippo4j.message.service.ThreadPoolNotifyAlarm;
 import cn.hippo4j.springboot.starter.config.BootstrapProperties;
+import cn.hippo4j.springboot.starter.core.DynamicThreadPoolSubscribeConfig;
 import cn.hippo4j.springboot.starter.remote.HttpAgent;
 import lombok.AllArgsConstructor;
 import lombok.extern.slf4j.Slf4j;