Merge pull request #284 from shining-stars-lk/develop

Hystrix is optimized for adaptation

hippo4j-adapter/hippo4j-adapter-base/src/main/java/cn/hippo4j/adapter/base/ThreadPoolAdapterScheduleAutoConfiguration.java

@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cn.hippo4j.adapter.base;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * thread pool adapter extra auto configuration.
+ */
+@Configuration(proxyBeanMethods = false)
+public class ThreadPoolAdapterScheduleAutoConfiguration {
+
+    @Bean
+    public ThreadPoolAdapterScheduler threadPoolAdapterExtra() {
+        return new ThreadPoolAdapterScheduler();
+    }
+
+}

hippo4j-adapter/hippo4j-adapter-base/src/main/java/cn/hippo4j/adapter/base/ThreadPoolAdapterScheduler.java

@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package cn.hippo4j.adapter.base;
+
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
+import lombok.extern.slf4j.Slf4j;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+
+/**
+ * thread pool adapter schedule.
+ */
+@Slf4j
+public class ThreadPoolAdapterScheduler {
+
+    private static final int TASK_INTERVAL_SECONDS = 2;
+
+    private final ScheduledExecutorService scheduler;
+
+
+    public ThreadPoolAdapterScheduler() {
+        scheduler = new ScheduledThreadPoolExecutor(2,
+                new ThreadFactoryBuilder()
+                        .setNameFormat("threadPoolAdapter")
+                        .setDaemon(true)
+                        .build());
+    }
+
+    public ScheduledExecutorService getScheduler() {
+        return scheduler;
+    }
+
+    public int getTaskIntervalSeconds(){
+        return TASK_INTERVAL_SECONDS;
+    }
+}

hippo4j-adapter/hippo4j-adapter-base/src/main/resources/META-INF/spring.factories

@@ -1 +1 @@
-org.springframework.boot.autoconfigure.EnableAutoConfiguration=cn.hippo4j.adapter.base.ThreadPoolAdapterExtraAutoConfiguration
+org.springframework.boot.autoconfigure.EnableAutoConfiguration=cn.hippo4j.adapter.base.ThreadPoolAdapterScheduleAutoConfiguration

hippo4j-adapter/hippo4j-adapter-hystrix/src/main/java/cn/hippo4j/adapter/hystrix/HystrixThreadPoolAdapter.java

@@ -18,13 +18,11 @@
 package cn.hippo4j.adapter.hystrix;
 
 import cn.hippo4j.adapter.base.ThreadPoolAdapter;
-import cn.hippo4j.adapter.base.ThreadPoolAdapterExtra;
+import cn.hippo4j.adapter.base.ThreadPoolAdapterScheduler;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterParameter;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
-import cn.hippo4j.common.config.ApplicationContextHolder;
 import cn.hippo4j.common.toolkit.CollectionUtil;
 import com.google.common.collect.Maps;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.netflix.hystrix.HystrixThreadPool;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
@@ -32,13 +30,10 @@ import org.springframework.context.ApplicationListener;
 
 import java.lang.reflect.Field;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.ScheduledThreadPoolExecutor;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 
@@ -54,23 +49,14 @@ public class HystrixThreadPoolAdapter implements ThreadPoolAdapter, ApplicationL
 
     private static final String THREAD_POOLS_FIELD = "threadPools";
 
-    private static final int TASK_INTERVAL_SECONDS = 2;
-
     private final Map<String, ThreadPoolExecutor> HYSTRIX_CONSUME_EXECUTOR = Maps.newHashMap();
 
-    private final ScheduledExecutorService scheduler;
-
-    private ThreadPoolAdapterExtra threadPoolAdapterExtra;
+    private ThreadPoolAdapterScheduler threadPoolAdapterScheduler;
 
-    public HystrixThreadPoolAdapter(ThreadPoolAdapterExtra threadPoolAdapterExtra) {
+    public HystrixThreadPoolAdapter(ThreadPoolAdapterScheduler threadPoolAdapterScheduler) {
 
-        this.threadPoolAdapterExtra = threadPoolAdapterExtra;
+        this.threadPoolAdapterScheduler = threadPoolAdapterScheduler;
 
-        scheduler = new ScheduledThreadPoolExecutor(2,
-                new ThreadFactoryBuilder()
-                        .setNameFormat("hystrixThreadPoolAdapter")
-                        .setDaemon(true)
-                        .build());
     }
 
     @Override
@@ -120,13 +106,14 @@ public class HystrixThreadPoolAdapter implements ThreadPoolAdapter, ApplicationL
 
     @Override
     public void onApplicationEvent(ApplicationStartedEvent event) {
-        HystrixThreadPoolRefreshTask hystrixThreadPoolRefreshTask = new HystrixThreadPoolRefreshTask(scheduler);
-        scheduler.schedule(hystrixThreadPoolRefreshTask, TASK_INTERVAL_SECONDS, TimeUnit.SECONDS);
+        ScheduledExecutorService scheduler = threadPoolAdapterScheduler.getScheduler();
+        int taskIntervalSeconds = threadPoolAdapterScheduler.getTaskIntervalSeconds();
+        HystrixThreadPoolRefreshTask hystrixThreadPoolRefreshTask = new HystrixThreadPoolRefreshTask(scheduler,taskIntervalSeconds);
+        scheduler.schedule(hystrixThreadPoolRefreshTask, taskIntervalSeconds, TimeUnit.SECONDS);
     }
 
     public void hystrixThreadPoolRefresh() {
         try {
-            boolean addExtraFlag = false;
             Class<HystrixThreadPool.Factory> factoryClass = HystrixThreadPool.Factory.class;
             Field threadPoolsField = factoryClass.getDeclaredField(THREAD_POOLS_FIELD);
             threadPoolsField.setAccessible(true);
@@ -144,18 +131,10 @@ public class HystrixThreadPoolAdapter implements ThreadPoolAdapter, ApplicationL
                         threadPoolField.setAccessible(true);
                         ThreadPoolExecutor threadPoolExecutor =
                                 (ThreadPoolExecutor) threadPoolField.get(hystrixThreadPoolDefault);
-                        if (threadPoolExecutor != null && HYSTRIX_CONSUME_EXECUTOR.get(key) == null) {
                         HYSTRIX_CONSUME_EXECUTOR.put(key, threadPoolExecutor);
-                            addExtraFlag = true;
-                        }
                     }
                 }
             }
-            if (addExtraFlag) {
-                Map<String, ThreadPoolAdapter> map = Maps.newHashMap();
-                map.putAll(ApplicationContextHolder.getBeansOfType(HystrixThreadPoolAdapter.class));
-                threadPoolAdapterExtra.offerQueue(map);
-            }
         } catch (Exception e) {
             log.error("Failed to get Hystrix thread pool.", e);
         }
@@ -166,8 +145,11 @@ public class HystrixThreadPoolAdapter implements ThreadPoolAdapter, ApplicationL
 
         private ScheduledExecutorService scheduler;
 
-        public HystrixThreadPoolRefreshTask(ScheduledExecutorService scheduler) {
+        private int taskIntervalSeconds;
+
+        public HystrixThreadPoolRefreshTask(ScheduledExecutorService scheduler, int taskIntervalSeconds) {
             this.scheduler = scheduler;
+            this.taskIntervalSeconds = taskIntervalSeconds;
         }
 
         @Override
@@ -176,7 +158,7 @@ public class HystrixThreadPoolAdapter implements ThreadPoolAdapter, ApplicationL
                 hystrixThreadPoolRefresh();
             } finally {
                 if (!scheduler.isShutdown()) {
-                    scheduler.schedule(this, TASK_INTERVAL_SECONDS, TimeUnit.MILLISECONDS);
+                    scheduler.schedule(this, taskIntervalSeconds, TimeUnit.MILLISECONDS);
                 }
             }
         }

hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-hystrix/src/main/java/cn/hippo4j/springboot/starter/adapter/hystrix/HystrixAdapterAutoConfiguration.java

@@ -17,8 +17,8 @@
 
 package cn.hippo4j.springboot.starter.adapter.hystrix;
 
-import cn.hippo4j.adapter.base.ThreadPoolAdapterExtra;
-import cn.hippo4j.adapter.base.ThreadPoolAdapterExtraAutoConfiguration;
+import cn.hippo4j.adapter.base.ThreadPoolAdapterScheduler;
+import cn.hippo4j.adapter.base.ThreadPoolAdapterScheduleAutoConfiguration;
 import cn.hippo4j.adapter.hystrix.HystrixThreadPoolAdapter;
 import cn.hippo4j.common.config.ApplicationContextHolder;
 import org.springframework.boot.autoconfigure.AutoConfigureAfter;
@@ -33,7 +33,7 @@ import org.springframework.context.annotation.Configuration;
  * @create: 2022-07-15
  **/
 @Configuration(proxyBeanMethods = false)
-@AutoConfigureAfter(ThreadPoolAdapterExtraAutoConfiguration.class)
+@AutoConfigureAfter(ThreadPoolAdapterScheduleAutoConfiguration.class)
 public class HystrixAdapterAutoConfiguration {
 
     @Bean
@@ -43,7 +43,7 @@ public class HystrixAdapterAutoConfiguration {
     }
 
     @Bean
-    public HystrixThreadPoolAdapter hystrixThreadPoolAdapter(ThreadPoolAdapterExtra threadPoolAdapterExtra) {
-        return new HystrixThreadPoolAdapter(threadPoolAdapterExtra);
+    public HystrixThreadPoolAdapter hystrixThreadPoolAdapter(ThreadPoolAdapterScheduler threadPoolAdapterScheduler) {
+        return new HystrixThreadPoolAdapter(threadPoolAdapterScheduler);
     }
 }

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

@@ -18,7 +18,7 @@
 package cn.hippo4j.springboot.starter.config;
 
 import cn.hippo4j.adapter.base.ThreadPoolAdapterBeanContainer;
-import cn.hippo4j.adapter.base.ThreadPoolAdapterExtra;
+import cn.hippo4j.adapter.base.ThreadPoolAdapterScheduler;
 import cn.hippo4j.common.api.ThreadDetailState;
 import cn.hippo4j.common.config.ApplicationContextHolder;
 import cn.hippo4j.core.config.UtilAutoConfiguration;
@@ -166,7 +166,7 @@ public class DynamicThreadPoolAutoConfiguration {
 
     @Bean
     @SuppressWarnings("all")
-    public ThreadPoolAdapterRegister threadPoolAdapterRegister(HttpAgent httpAgent, InetUtils hippo4JInetUtils, ThreadPoolAdapterExtra threadPoolAdapterExtra) {
-        return new ThreadPoolAdapterRegister(httpAgent, properties, environment, hippo4JInetUtils, threadPoolAdapterExtra);
+    public ThreadPoolAdapterRegister threadPoolAdapterRegister(HttpAgent httpAgent, InetUtils hippo4JInetUtils, ThreadPoolAdapterScheduler threadPoolAdapterScheduler) {
+        return new ThreadPoolAdapterRegister(httpAgent, properties, environment, hippo4JInetUtils, threadPoolAdapterScheduler);
     }
 }

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

@@ -19,7 +19,7 @@ package cn.hippo4j.springboot.starter.core;
 
 import cn.hippo4j.adapter.base.ThreadPoolAdapter;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterCacheConfig;
-import cn.hippo4j.adapter.base.ThreadPoolAdapterExtra;
+import cn.hippo4j.adapter.base.ThreadPoolAdapterScheduler;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
 import cn.hippo4j.common.config.ApplicationContextHolder;
 import cn.hippo4j.common.toolkit.CollectionUtil;
@@ -31,6 +31,7 @@ import cn.hippo4j.springboot.starter.remote.HttpAgent;
 import cn.hippo4j.springboot.starter.toolkit.CloudCommonIdUtil;
 import com.google.common.collect.Lists;
 import lombok.AllArgsConstructor;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
@@ -38,6 +39,9 @@ import org.springframework.core.env.ConfigurableEnvironment;
 
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Collectors;
 
 import static cn.hippo4j.common.constant.Constants.IDENTIFY_SLICER_SYMBOL;
 import static cn.hippo4j.common.constant.Constants.REGISTER_ADAPTER_PATH;
@@ -47,6 +51,7 @@ import static cn.hippo4j.common.constant.Constants.REGISTER_ADAPTER_PATH;
  */
 @Slf4j
 @AllArgsConstructor
+@RequiredArgsConstructor
 public class ThreadPoolAdapterRegister implements ApplicationRunner {
 
     private final HttpAgent httpAgent;
@@ -57,17 +62,23 @@ public class ThreadPoolAdapterRegister implements ApplicationRunner {
 
     private final InetUtils hippo4JInetUtils;
 
-    private final ThreadPoolAdapterExtra threadPoolAdapterExtra;
+    private final ThreadPoolAdapterScheduler threadPoolAdapterScheduler;
+
+    private List<ThreadPoolAdapterCacheConfig> cacheConfigList = Lists.newArrayList();
+
+
 
     @Override
     public void run(ApplicationArguments args) throws Exception {
-        Map<String, ThreadPoolAdapter> threadPoolAdapterMap = ApplicationContextHolder.getBeansOfType(ThreadPoolAdapter.class);
-        register(threadPoolAdapterMap);
-        threadPoolAdapterExtra.extraStart(map -> register(map));
-    }
 
-    public void register(Map<String, ThreadPoolAdapter> threadPoolAdapterMap) {
+        ScheduledExecutorService scheduler = threadPoolAdapterScheduler.getScheduler();
+        int taskIntervalSeconds = threadPoolAdapterScheduler.getTaskIntervalSeconds();
+        ThreadPoolAdapterRegisterTask threadPoolAdapterRegisterTask = new ThreadPoolAdapterRegisterTask(scheduler, taskIntervalSeconds);
+        scheduler.schedule(threadPoolAdapterRegisterTask, threadPoolAdapterScheduler.getTaskIntervalSeconds(), TimeUnit.SECONDS);
+    }
 
+    public List<ThreadPoolAdapterCacheConfig> getThreadPoolAdapterCacheConfigs(){
+        Map<String, ThreadPoolAdapter> threadPoolAdapterMap = ApplicationContextHolder.getBeansOfType(ThreadPoolAdapter.class);
         List<ThreadPoolAdapterCacheConfig> cacheConfigList = Lists.newArrayList();
         threadPoolAdapterMap.forEach((key, val) -> {
             List<ThreadPoolAdapterState> threadPoolStates = val.getThreadPoolStates();
@@ -84,6 +95,10 @@ public class ThreadPoolAdapterRegister implements ApplicationRunner {
             cacheConfig.setThreadPoolAdapterStates(threadPoolStates);
             cacheConfigList.add(cacheConfig);
         });
+        return cacheConfigList;
+    }
+
+    public void doRegister(List<ThreadPoolAdapterCacheConfig> cacheConfigList){
         if (CollectionUtil.isNotEmpty(cacheConfigList)) {
             try {
                 Result result = httpAgent.httpPost(REGISTER_ADAPTER_PATH, cacheConfigList);
@@ -95,4 +110,71 @@ public class ThreadPoolAdapterRegister implements ApplicationRunner {
             }
         }
     }
+
+    public void register() {
+        List<ThreadPoolAdapterCacheConfig> threadPoolAdapterCacheConfigs = getThreadPoolAdapterCacheConfigs();
+        doRegister(threadPoolAdapterCacheConfigs);
+    }
+
+    class ThreadPoolAdapterRegisterTask implements Runnable{
+
+        private ScheduledExecutorService scheduler;
+
+        private int taskIntervalSeconds;
+
+        public ThreadPoolAdapterRegisterTask(ScheduledExecutorService scheduler, int taskIntervalSeconds){
+            this.scheduler = scheduler;
+            this.taskIntervalSeconds = taskIntervalSeconds;
+        }
+
+        @Override
+        public void run() {
+            try {
+                List<ThreadPoolAdapterCacheConfig> newThreadPoolAdapterCacheConfigs = getThreadPoolAdapterCacheConfigs();
+
+                boolean registerFlag = compareThreadPoolAdapterCacheConfigs(newThreadPoolAdapterCacheConfigs, cacheConfigList);
+
+                cacheConfigList = newThreadPoolAdapterCacheConfigs;
+
+                if (registerFlag) {
+                    doRegister(cacheConfigList);
+                }
+            }catch (Exception e){
+                log.error("Register Task Error",e);
+            }finally {
+                if (!scheduler.isShutdown()) {
+                    scheduler.schedule(this, taskIntervalSeconds, TimeUnit.MILLISECONDS);
+                }
+            }
+        }
+    }
+
+    private boolean compareThreadPoolAdapterCacheConfigs(List<ThreadPoolAdapterCacheConfig> newThreadPoolAdapterCacheConfigs,
+                                                         List<ThreadPoolAdapterCacheConfig> oldThreadPoolAdapterCacheConfigs){
+        boolean registerFlag = false;
+
+        Map<String, List<ThreadPoolAdapterState>> newThreadPoolAdapterCacheConfigMap =
+                newThreadPoolAdapterCacheConfigs.stream().collect(Collectors.toMap(
+                        ThreadPoolAdapterCacheConfig::getMark, ThreadPoolAdapterCacheConfig::getThreadPoolAdapterStates, (k1, k2) -> k2));
+
+        Map<String, List<ThreadPoolAdapterState>> oldThreadPoolAdapterCacheConfigMap =
+                oldThreadPoolAdapterCacheConfigs.stream().collect(Collectors.toMap(
+                        ThreadPoolAdapterCacheConfig::getMark, ThreadPoolAdapterCacheConfig::getThreadPoolAdapterStates, (k1, k2) -> k2));
+
+        for (Map.Entry<String, List<ThreadPoolAdapterState>> entry : newThreadPoolAdapterCacheConfigMap.entrySet()) {
+            String key = entry.getKey();
+            List<ThreadPoolAdapterState> newValue = entry.getValue();
+            List<ThreadPoolAdapterState> oldValue = oldThreadPoolAdapterCacheConfigMap.get(key);
+            if (oldValue == null) {
+                registerFlag = true;
+                break;
+            }else {
+                if (newValue.size() != oldValue.size()) {
+                    registerFlag = true;
+                    break;
+                }
+            }
+        }
+        return registerFlag;
+    }
 }