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

Support rabbitmq multi-connection (#541)

Browse Source
pull/546/head^2
weihubeats 2 years ago committed by GitHub
parent 0795538f07
commit a956bc0d64
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 81 additions and 78 deletions
Show Stats Download Patch File Download Diff File

152
hippo4j-adapter/hippo4j-adapter-rabbitmq/src/main/java/cn/hippo4j/adapter/rabbitmq/RabbitMQThreadPoolAdapter.java
Unescape View File

@ -17,6 +17,12 @@
package cn.hippo4j.adapter.rabbitmq; package cn.hippo4j.adapter.rabbitmq;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import cn.hippo4j.adapter.base.ThreadPoolAdapter; import cn.hippo4j.adapter.base.ThreadPoolAdapter;
import cn.hippo4j.adapter.base.ThreadPoolAdapterParameter; import cn.hippo4j.adapter.base.ThreadPoolAdapterParameter;
import cn.hippo4j.adapter.base.ThreadPoolAdapterState; import cn.hippo4j.adapter.base.ThreadPoolAdapterState;
@ -25,16 +31,11 @@ import com.google.common.collect.Lists;
import com.google.common.collect.Maps; import com.google.common.collect.Maps;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j; import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory; import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.boot.context.event.ApplicationStartedEvent; import org.springframework.boot.context.event.ApplicationStartedEvent;
import org.springframework.context.ApplicationListener; import org.springframework.context.ApplicationListener;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER; import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
/** /**
@ -44,74 +45,73 @@ import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMI
@RequiredArgsConstructor @RequiredArgsConstructor
public class RabbitMQThreadPoolAdapter implements ThreadPoolAdapter, ApplicationListener<ApplicationStartedEvent> { public class RabbitMQThreadPoolAdapter implements ThreadPoolAdapter, ApplicationListener<ApplicationStartedEvent> {
private static final String RABBITMQ = "RabbitMQ"; private static final String RABBITMQ = "RabbitMQ";
private static final String FiledName = "executorService"; private static final String FiledName = "executorService";
/** private final Map<String, AbstractConnectionFactory> abstractConnectionFactoryMap;
* TODO Configurable name
*/ private final Map<String, ThreadPoolExecutor> RABBITMQ_THREAD_POOL_TASK_EXECUTOR = Maps.newHashMap();
private static final String RABBITMQ_EXECUTOR_SERVICE = "Rabbitmq_Executor_Service";
@Override
private final AbstractConnectionFactory abstractConnectionFactory; public String mark() {
return RABBITMQ;
private final Map<String, ThreadPoolExecutor> RABBITMQ_THREAD_POOL_TASK_EXECUTOR = Maps.newHashMap(); }
@Override @Override
public String mark() { public ThreadPoolAdapterState getThreadPoolState(String identify) {
return RABBITMQ; ThreadPoolAdapterState threadPoolAdapterState = new ThreadPoolAdapterState();
} ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(identify);
threadPoolAdapterState.setThreadPoolKey(identify);
@Override if (Objects.nonNull(threadPoolTaskExecutor)) {
public ThreadPoolAdapterState getThreadPoolState(String identify) { threadPoolAdapterState.setCoreSize(threadPoolTaskExecutor.getCorePoolSize());
ThreadPoolAdapterState threadPoolAdapterState = new ThreadPoolAdapterState(); threadPoolAdapterState.setMaximumSize(threadPoolTaskExecutor.getMaximumPoolSize());
ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(identify); }
threadPoolAdapterState.setThreadPoolKey(identify); return threadPoolAdapterState;
if (Objects.nonNull(threadPoolTaskExecutor)) { }
threadPoolAdapterState.setCoreSize(threadPoolTaskExecutor.getCorePoolSize());
threadPoolAdapterState.setMaximumSize(threadPoolTaskExecutor.getMaximumPoolSize()); @Override
} public List<ThreadPoolAdapterState> getThreadPoolStates() {
return threadPoolAdapterState; List<ThreadPoolAdapterState> adapterStateList = Lists.newArrayList();
} RABBITMQ_THREAD_POOL_TASK_EXECUTOR.forEach(
(key, val) -> adapterStateList.add(getThreadPoolState(key)));
@Override return adapterStateList;
public List<ThreadPoolAdapterState> getThreadPoolStates() { }
List<ThreadPoolAdapterState> adapterStateList = Lists.newArrayList();
RABBITMQ_THREAD_POOL_TASK_EXECUTOR.forEach( @Override
(key, val) -> adapterStateList.add(getThreadPoolState(key))); public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) {
return adapterStateList; String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey();
} ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(threadPoolKey);
if (Objects.nonNull(threadPoolTaskExecutor)) {
@Override int originalCoreSize = threadPoolTaskExecutor.getCorePoolSize();
public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) { int originalMaximumPoolSize = threadPoolTaskExecutor.getMaximumPoolSize();
String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey(); threadPoolTaskExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize());
ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(threadPoolKey); threadPoolTaskExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize());
if (Objects.nonNull(threadPoolTaskExecutor)) { log.info("[{}] Rabbitmq consumption thread pool parameter change. coreSize: {}, maximumSize: {}",
int originalCoreSize = threadPoolTaskExecutor.getCorePoolSize(); threadPoolKey,
int originalMaximumPoolSize = threadPoolTaskExecutor.getMaximumPoolSize(); String.format(CHANGE_DELIMITER, originalCoreSize, threadPoolAdapterParameter.getCorePoolSize()),
threadPoolTaskExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize()); String.format(CHANGE_DELIMITER, originalMaximumPoolSize, threadPoolAdapterParameter.getMaximumPoolSize()));
threadPoolTaskExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize()); return true;
log.info("[{}] Rabbitmq consumption thread pool parameter change. coreSize: {}, maximumSize: {}", }
threadPoolKey, log.warn("[{}] Rabbitmq consuming thread pool not found.", threadPoolKey);
String.format(CHANGE_DELIMITER, originalCoreSize, threadPoolAdapterParameter.getCorePoolSize()), return false;
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, threadPoolAdapterParameter.getMaximumPoolSize())); }
return true;
} @Override
log.warn("[{}] Rabbitmq consuming thread pool not found.", threadPoolKey); public void onApplicationEvent(ApplicationStartedEvent event) {
return false; abstractConnectionFactoryMap.forEach((beanName, abstractConnectionFactor) -> {
} ExecutorService executor = (ExecutorService) ReflectUtil.getFieldValue(abstractConnectionFactor, FiledName);
if (Objects.nonNull(executor)) {
@Override if (executor instanceof ThreadPoolExecutor) {
public void onApplicationEvent(ApplicationStartedEvent event) { ThreadPoolExecutor threadPoolTaskExecutor = (ThreadPoolExecutor) executor;
ExecutorService executor = (ExecutorService) ReflectUtil.getFieldValue(abstractConnectionFactory, FiledName); RABBITMQ_THREAD_POOL_TASK_EXECUTOR.put(beanName, threadPoolTaskExecutor);
if (Objects.nonNull(executor)) { log.info("Rabbitmq executor name {}", beanName);
if (executor instanceof ThreadPoolExecutor) { }
ThreadPoolExecutor threadPoolTaskExecutor = (ThreadPoolExecutor) executor; else {
RABBITMQ_THREAD_POOL_TASK_EXECUTOR.put(RABBITMQ_EXECUTOR_SERVICE, threadPoolTaskExecutor); log.warn("Custom thread pools only support ThreadPoolExecutor");
log.info("Rabbitmq executor name {}", RABBITMQ_EXECUTOR_SERVICE); }
} else { }
log.warn("Custom thread pools only support ThreadPoolExecutor");
} });
} }
}
} }

7
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
Unescape View File

@ -17,9 +17,12 @@
package cn.hippo4j.springboot.starter.adapter.rabbitmq; package cn.hippo4j.springboot.starter.adapter.rabbitmq;
import java.util.Map;
import cn.hippo4j.adapter.rabbitmq.RabbitMQThreadPoolAdapter; import cn.hippo4j.adapter.rabbitmq.RabbitMQThreadPoolAdapter;
import cn.hippo4j.common.config.ApplicationContextHolder; import cn.hippo4j.common.config.ApplicationContextHolder;
import lombok.RequiredArgsConstructor; import lombok.RequiredArgsConstructor;
import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory; import org.springframework.amqp.rabbit.connection.AbstractConnectionFactory;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean; import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@ -33,7 +36,7 @@ import org.springframework.context.annotation.Configuration;
@RequiredArgsConstructor @RequiredArgsConstructor
public class RabbitMQAdapterAutoConfiguration { public class RabbitMQAdapterAutoConfiguration {
private final AbstractConnectionFactory abstractConnectionFactories; private final Map<String, AbstractConnectionFactory> stringAbstractConnectionFactoryMap;
@Bean @Bean
@ConditionalOnMissingBean @ConditionalOnMissingBean
@ -45,6 +48,6 @@ public class RabbitMQAdapterAutoConfiguration {
@SuppressWarnings("all") @SuppressWarnings("all")
@ConditionalOnProperty(name = "spring.rabbitmq.host") @ConditionalOnProperty(name = "spring.rabbitmq.host")
public RabbitMQThreadPoolAdapter rabbitMQThreadPoolAdapter(ApplicationContextHolder applicationContextHolder) { public RabbitMQThreadPoolAdapter rabbitMQThreadPoolAdapter(ApplicationContextHolder applicationContextHolder) {
return new RabbitMQThreadPoolAdapter(abstractConnectionFactories); return new RabbitMQThreadPoolAdapter(stringAbstractConnectionFactoryMap);
} }
} }

Write Preview
Loading…
Cancel
Save