Merge pull request #244 from weihubeats/feature_rabbitmq

Feature rabbitmq

hippo4j-adapter/hippo4j-adapter-rabbitmq/pom.xml

@@ -14,6 +14,11 @@
             <groupId>cn.hippo4j</groupId>
             <artifactId>hippo4j-adapter-base</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-amqp</artifactId>
+        </dependency>
+
     </dependencies>
 
     <build>

hippo4j-adapter/hippo4j-adapter-rabbitmq/src/main/java/cn/hippo4j/adapter/rabbitmq/RabbitMQThreadPoolAdapter.java

@@ -20,33 +20,113 @@ package cn.hippo4j.adapter.rabbitmq;
 import cn.hippo4j.adapter.base.ThreadPoolAdapter;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterParameter;
 import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
+import cn.hippo4j.common.toolkit.ReflectUtil;
+import com.google.common.collect.Maps;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.listener.AbstractMessageListenerContainer;
 import org.springframework.boot.context.event.ApplicationStartedEvent;
 import org.springframework.context.ApplicationListener;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.concurrent.Executor;
+
+import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
 
 /**
  * RabbitMQ thread-pool adapter.
  */
 @Slf4j
+@RequiredArgsConstructor
 public class RabbitMQThreadPoolAdapter implements ThreadPoolAdapter, ApplicationListener<ApplicationStartedEvent> {
 
+    private static final String RABBITMQ = "RabbitMQ";
+
+    private static final String FiledName = "taskExecutor";
+
+    private static final String BEAN_NAME_FILED = "beanName";
+
+    private final List<AbstractRabbitListenerContainerFactory<?>> abstractRabbitListenerContainerFactories;
+
+    private final Map<String, SimpleAsyncTaskExecutor> RABBITMQ_EXECUTOR = Maps.newHashMap();
+
+    private final Map<String, ThreadPoolTaskExecutor> RABBITMQ_THREAD_POOL_TASK_EXECUTOR = Maps.newHashMap();
+
     @Override
     public String mark() {
-        return "RabbitMQ";
+        return RABBITMQ;
     }
 
     @Override
     public ThreadPoolAdapterState getThreadPoolState(String identify) {
-        return null;
+        ThreadPoolAdapterState threadPoolAdapterState = new ThreadPoolAdapterState();
+        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = RABBITMQ_EXECUTOR.get(identify);
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(identify);
+        threadPoolAdapterState.setThreadPoolKey(identify);
+        if (Objects.nonNull(simpleAsyncTaskExecutor)) {
+            threadPoolAdapterState.setCoreSize(simpleAsyncTaskExecutor.getConcurrencyLimit());
+        }
+        if (Objects.nonNull(threadPoolTaskExecutor)) {
+            threadPoolAdapterState.setCoreSize(threadPoolTaskExecutor.getCorePoolSize());
+            threadPoolAdapterState.setMaximumSize(threadPoolTaskExecutor.getMaxPoolSize());
+        }
+        return threadPoolAdapterState;
     }
 
     @Override
     public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) {
+        String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey();
+        SimpleAsyncTaskExecutor simpleAsyncTaskExecutor = RABBITMQ_EXECUTOR.get(threadPoolKey);
+        ThreadPoolTaskExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(threadPoolKey);
+        if (Objects.nonNull(threadPoolTaskExecutor)) {
+            int originalCoreSize = threadPoolTaskExecutor.getCorePoolSize();
+            int originalMaximumPoolSize = threadPoolTaskExecutor.getMaxPoolSize();
+            threadPoolTaskExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize());
+            threadPoolTaskExecutor.setMaxPoolSize(threadPoolAdapterParameter.getMaximumPoolSize());
+            log.info("[{}] rabbitmq consumption thread pool parameter change. coreSize :: {}, maximumSize :: {}",
+                    threadPoolKey,
+                    String.format(CHANGE_DELIMITER, originalCoreSize, threadPoolAdapterParameter.getCorePoolSize()),
+                    String.format(CHANGE_DELIMITER, originalMaximumPoolSize, threadPoolAdapterParameter.getMaximumPoolSize()));
+            return true;
+        }
+        if (Objects.nonNull(simpleAsyncTaskExecutor)) {
+            int concurrencyLimit = simpleAsyncTaskExecutor.getConcurrencyLimit();
+            simpleAsyncTaskExecutor.setConcurrencyLimit(threadPoolAdapterParameter.getCorePoolSize());
+            log.info("[{}] rabbitmq consumption thread pool parameter change. coreSize :: {}",
+                    threadPoolKey,
+                    String.format(CHANGE_DELIMITER, concurrencyLimit, threadPoolAdapterParameter.getCorePoolSize()));
+            return true;
+        }
+        log.warn("[{}] rabbitmq consuming thread pool not found.", threadPoolKey);
         return false;
     }
 
     @Override
     public void onApplicationEvent(ApplicationStartedEvent event) {
+        for (AbstractRabbitListenerContainerFactory<?> consumerWorkService : abstractRabbitListenerContainerFactories) {
+            // 是否为自定义线程池
+            Executor executor = (Executor) ReflectUtil.getFieldValue(consumerWorkService, FiledName);
+            if (Objects.isNull(executor)) {
+                // 获取默认线程池
+                // 优先获取用户配置的
+                AbstractMessageListenerContainer listenerContainer1 = consumerWorkService.createListenerContainer();
+                SimpleAsyncTaskExecutor fieldValue = (SimpleAsyncTaskExecutor) ReflectUtil.getFieldValue(listenerContainer1, FiledName);
+                RABBITMQ_EXECUTOR.put(FiledName, fieldValue);
+            } else {
+                if (executor instanceof ThreadPoolTaskExecutor) {
+                    ThreadPoolTaskExecutor threadPoolTaskExecutor = (ThreadPoolTaskExecutor) executor;
+                    String beanName = (String) ReflectUtil.getFieldValue(threadPoolTaskExecutor, BEAN_NAME_FILED);
+                    RABBITMQ_THREAD_POOL_TASK_EXECUTOR.put(beanName, threadPoolTaskExecutor);
+                } else {
+                    log.warn("Custom thread pools only support ThreadPoolTaskExecutor");
+                }
+            }
+        }
 
     }
 }

hippo4j-example/hippo4j-spring-boot-starter-adapter-rabbitmq-example/src/main/java/cn/hippo4j/springboot/starter/adapter/rabbitmq/example/config/RabbitMQThreadPoolConfig.java

@@ -0,0 +1,44 @@
+package cn.hippo4j.springboot.starter.adapter.rabbitmq.example.config;
+
+import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+/**
+ * @author : wh
+ * @date : 2022/5/24 10:02
+ * @description:
+ */
+@Configuration
+public class RabbitMQThreadPoolConfig {
+
+    
+    @Bean
+    public ThreadPoolTaskExecutor rabbitListenerTaskExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        executor.setMaxPoolSize(10); // 指定线程的最大数量
+        executor.setCorePoolSize(2); // 指定线程池维护线程的最少数量
+        executor.setQueueCapacity(20); // 指定等待处理的任务数
+        executor.setThreadNamePrefix("RabbitListenerTaskExecutor-");
+        return executor;
+    }
+
+    @Bean
+    public AbstractRabbitListenerContainerFactory<?> defaultRabbitListenerContainerFactory (
+            SimpleRabbitListenerContainerFactoryConfigurer configurer,
+            ThreadPoolTaskExecutor rabbitListenerTaskExecutor,
+            ConnectionFactory connectionFactory) {
+        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
+        configurer.configure(factory, connectionFactory);
+        factory.setConcurrentConsumers(1);
+        factory.setMaxConcurrentConsumers(10);
+        factory.setTaskExecutor(rabbitListenerTaskExecutor);
+        return factory;
+    }
+
+
+}

hippo4j-spring-boot/hippo4j-spring-boot-starter-adapter/hippo4j-spring-boot-starter-adapter-rabbitmq/src/main/java/cn/hippo4j/springboot/starter/adapter/rabbitmq/RabbitMQAdapterAutoConfiguration.java

@@ -19,17 +19,24 @@ package cn.hippo4j.springboot.starter.adapter.rabbitmq;
 
 import cn.hippo4j.adapter.rabbitmq.RabbitMQThreadPoolAdapter;
 import cn.hippo4j.common.config.ApplicationContextHolder;
+import lombok.RequiredArgsConstructor;
+import org.springframework.amqp.rabbit.config.AbstractRabbitListenerContainerFactory;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
+import java.util.List;
+
 /**
  * Rabbit adapter auto configuration.
  */
 @Configuration
+@RequiredArgsConstructor
 public class RabbitMQAdapterAutoConfiguration {
 
+    private final List<AbstractRabbitListenerContainerFactory<?>> abstractRabbitListenerContainerFactories;
+
     @Bean
     @ConditionalOnMissingBean
     public ApplicationContextHolder simpleApplicationContextHolder() {
@@ -38,8 +45,8 @@ public class RabbitMQAdapterAutoConfiguration {
 
     @Bean
     @SuppressWarnings("all")
-    @ConditionalOnProperty(name = "spring.rabbitmq.addresses")
+    @ConditionalOnProperty(name = "spring.rabbitmq.host")
     public RabbitMQThreadPoolAdapter rabbitMQThreadPoolAdapter(ApplicationContextHolder applicationContextHolder) {
-        return new RabbitMQThreadPoolAdapter();
+        return new RabbitMQThreadPoolAdapter(abstractRabbitListenerContainerFactories);
     }
 }