Merge pull request #505 from mabaiwan/develop

Thread pool dynamic registration optimization

hippo4j-common/src/main/java/cn/hippo4j/common/toolkit/ReflectUtil.java

@@ -17,6 +17,10 @@
 
 package cn.hippo4j.common.toolkit;
 
+import cn.hutool.core.convert.Convert;
+import cn.hutool.core.exceptions.UtilException;
+import cn.hutool.core.util.ClassUtil;
+
 import java.lang.reflect.AccessibleObject;
 import java.lang.reflect.Field;
 import java.util.Arrays;
@@ -106,4 +110,33 @@ public class ReflectUtil {
         }
         return field.getName();
     }
+
+    public static void setFieldValue(Object obj, String fieldName, Object value) throws UtilException {
+        cn.hutool.core.lang.Assert.notNull(obj);
+        cn.hutool.core.lang.Assert.notBlank(fieldName);
+        final Field field = getField((obj instanceof Class) ? (Class<?>) obj : obj.getClass(), fieldName);
+        cn.hutool.core.lang.Assert.notNull(field, "Field [{}] is not exist in [{}]", fieldName, obj.getClass().getName());
+        setFieldValue(obj, field, value);
+    }
+
+    public static void setFieldValue(Object obj, Field field, Object value) throws UtilException {
+        cn.hutool.core.lang.Assert.notNull(field, "Field in [{}] not exist !", obj);
+        final Class<?> fieldType = field.getType();
+        if (null != value) {
+            if (false == fieldType.isAssignableFrom(value.getClass())) {
+                final Object targetValue = Convert.convert(fieldType, value);
+                if (null != targetValue) {
+                    value = targetValue;
+                }
+            }
+        } else {
+            value = ClassUtil.getDefaultValue(fieldType);
+        }
+        setAccessible(field);
+        try {
+            field.set(obj instanceof Class ? null : obj, value);
+        } catch (IllegalAccessException e) {
+            throw new UtilException(e, "IllegalAccess for {}.{}", obj, field.getName());
+        }
+    }
 }

hippo4j-core/src/main/java/cn/hippo4j/core/executor/manage/GlobalThreadPoolManage.java

@@ -17,8 +17,11 @@
 
 package cn.hippo4j.core.executor.manage;
 
+import cn.hippo4j.common.config.ApplicationContextHolder;
 import cn.hippo4j.common.model.ThreadPoolParameter;
+import cn.hippo4j.common.model.register.DynamicThreadPoolRegisterWrapper;
 import cn.hippo4j.core.executor.DynamicThreadPoolWrapper;
+import cn.hippo4j.core.executor.support.DynamicThreadPoolService;
 import com.google.common.collect.Lists;
 
 import java.util.List;
@@ -104,6 +107,16 @@ public class GlobalThreadPoolManage {
         POOL_PARAMETER.put(threadPoolId, poolParameter);
     }
 
+    /**
+     * Dynamically register thread pool records and notification records.
+     *
+     * @param registerWrapper
+     */
+    public static ThreadPoolExecutor dynamicRegister(DynamicThreadPoolRegisterWrapper registerWrapper) {
+        DynamicThreadPoolService dynamicThreadPoolService = ApplicationContextHolder.getBean(DynamicThreadPoolService.class);
+        return dynamicThreadPoolService.registerDynamicThreadPool(registerWrapper);
+    }
+
     /**
      * Get the dynamic thread pool identifier collection.
      *

hippo4j-example/hippo4j-example-core/pom.xml

@@ -29,5 +29,10 @@
             <groupId>cn.hippo4j</groupId>
             <artifactId>hippo4j-core</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>transmittable-thread-local</artifactId>
+        </dependency>
     </dependencies>
 </project>

hippo4j-example/hippo4j-example-core/src/main/java/cn/hippo4j/example/core/config/ThreadPoolConfig.java

@@ -21,10 +21,12 @@ import cn.hippo4j.core.executor.DynamicThreadPool;
 import cn.hippo4j.core.executor.support.ThreadPoolBuilder;
 import cn.hippo4j.example.core.handler.TaskTraceBuilderHandler;
 import cn.hippo4j.example.core.inittest.TaskDecoratorTest;
+import com.alibaba.ttl.threadpool.TtlExecutors;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.concurrent.Executor;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import static cn.hippo4j.example.core.constant.GlobalTestConstant.MESSAGE_CONSUME;
@@ -75,4 +77,37 @@ public class ThreadPoolConfig {
         return produceExecutor;
     }
 
+    @Bean
+    @DynamicThreadPool
+    public Executor messageConsumeTtlDynamicThreadPool() {
+        String threadPoolId = MESSAGE_CONSUME;
+        ThreadPoolExecutor customExecutor = ThreadPoolBuilder.builder()
+                .dynamicPool()
+                .threadFactory(threadPoolId)
+                .threadPoolId(threadPoolId)
+                .executeTimeOut(800L)
+                .waitForTasksToCompleteOnShutdown(true)
+                .awaitTerminationMillis(5000L)
+                .taskDecorator(new TaskTraceBuilderHandler())
+                .build();
+        Executor ttlExecutor = TtlExecutors.getTtlExecutor(customExecutor);
+        return ttlExecutor;
+    }
+
+    @Bean
+    @DynamicThreadPool
+    public Executor messageConsumeTtlServiceDynamicThreadPool() {
+        String threadPoolId = MESSAGE_CONSUME;
+        ThreadPoolExecutor customExecutor = ThreadPoolBuilder.builder()
+                .dynamicPool()
+                .threadFactory(threadPoolId)
+                .threadPoolId(threadPoolId)
+                .executeTimeOut(800L)
+                .waitForTasksToCompleteOnShutdown(true)
+                .awaitTerminationMillis(5000L)
+                .taskDecorator(new TaskTraceBuilderHandler())
+                .build();
+        Executor ttlExecutor = TtlExecutors.getTtlExecutorService(customExecutor);
+        return ttlExecutor;
+    }
 }

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

@@ -1,10 +1,26 @@
+/*
+ * 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.example.core.inittest;
 
 import cn.hippo4j.common.model.register.DynamicThreadPoolRegisterParameter;
 import cn.hippo4j.common.model.register.DynamicThreadPoolRegisterWrapper;
 import cn.hippo4j.common.toolkit.JSONUtil;
-import cn.hippo4j.core.executor.support.DynamicThreadPoolService;
-import lombok.AllArgsConstructor;
+import cn.hippo4j.core.executor.manage.GlobalThreadPoolManage;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
@@ -12,21 +28,18 @@ import javax.annotation.PostConstruct;
 import java.util.concurrent.ThreadPoolExecutor;
 
 /**
- * Register dynamic thread-pool test
+ * Register dynamic thread-pool test.
  */
 @Slf4j
 @Component
-@AllArgsConstructor
 public class RegisterDynamicThreadPoolTest {
 
-    private final DynamicThreadPoolService dynamicThreadPoolService;
-
     @PostConstruct
     public void registerDynamicThreadPool() {
         String threadPoolId = "register-dynamic-thread-pool";
         DynamicThreadPoolRegisterParameter parameterInfo = new DynamicThreadPoolRegisterParameter();
         parameterInfo.setThreadPoolId(threadPoolId);
-        parameterInfo.setCorePoolSize(2);
+        parameterInfo.setCorePoolSize(3);
         parameterInfo.setMaximumPoolSize(14);
         parameterInfo.setQueueType(9);
         parameterInfo.setCapacity(110);
@@ -39,7 +52,7 @@ public class RegisterDynamicThreadPoolTest {
         DynamicThreadPoolRegisterWrapper registerWrapper = DynamicThreadPoolRegisterWrapper.builder()
                 .dynamicThreadPoolRegisterParameter(parameterInfo)
                 .build();
-        ThreadPoolExecutor dynamicThreadPool = dynamicThreadPoolService.registerDynamicThreadPool(registerWrapper);
+        ThreadPoolExecutor dynamicThreadPool = GlobalThreadPoolManage.dynamicRegister(registerWrapper);
         log.info("Dynamic registration thread pool parameter details: {}", JSONUtil.toJSONString(dynamicThreadPool));
     }
 }