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

Code formatting optimization

Browse Source
pull/550/head
chen.ma 3 years ago
parent e86cbd641b
commit 7467ea634c
1 changed files with 74 additions and 76 deletions
Show Stats Download Patch File Download Diff File

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

@ -17,12 +17,6 @@
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;
@ -31,11 +25,16 @@ 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;
/** /**
@ -45,73 +44,72 @@ 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; private final Map<String, AbstractConnectionFactory> abstractConnectionFactoryMap;
private final Map<String, ThreadPoolExecutor> RABBITMQ_THREAD_POOL_TASK_EXECUTOR = Maps.newHashMap(); private final Map<String, ThreadPoolExecutor> RABBITMQ_THREAD_POOL_TASK_EXECUTOR = Maps.newHashMap();
@Override @Override
public String mark() { public String mark() {
return RABBITMQ; return RABBITMQ;
} }
@Override @Override
public ThreadPoolAdapterState getThreadPoolState(String identify) { public ThreadPoolAdapterState getThreadPoolState(String identify) {
ThreadPoolAdapterState threadPoolAdapterState = new ThreadPoolAdapterState(); ThreadPoolAdapterState threadPoolAdapterState = new ThreadPoolAdapterState();
ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(identify); ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(identify);
threadPoolAdapterState.setThreadPoolKey(identify); threadPoolAdapterState.setThreadPoolKey(identify);
if (Objects.nonNull(threadPoolTaskExecutor)) { if (Objects.nonNull(threadPoolTaskExecutor)) {
threadPoolAdapterState.setCoreSize(threadPoolTaskExecutor.getCorePoolSize()); threadPoolAdapterState.setCoreSize(threadPoolTaskExecutor.getCorePoolSize());
threadPoolAdapterState.setMaximumSize(threadPoolTaskExecutor.getMaximumPoolSize()); threadPoolAdapterState.setMaximumSize(threadPoolTaskExecutor.getMaximumPoolSize());
} }
return threadPoolAdapterState; return threadPoolAdapterState;
} }
@Override @Override
public List<ThreadPoolAdapterState> getThreadPoolStates() { public List<ThreadPoolAdapterState> getThreadPoolStates() {
List<ThreadPoolAdapterState> adapterStateList = Lists.newArrayList(); List<ThreadPoolAdapterState> adapterStateList = Lists.newArrayList();
RABBITMQ_THREAD_POOL_TASK_EXECUTOR.forEach( RABBITMQ_THREAD_POOL_TASK_EXECUTOR.forEach(
(key, val) -> adapterStateList.add(getThreadPoolState(key))); (key, val) -> adapterStateList.add(getThreadPoolState(key)));
return adapterStateList; return adapterStateList;
} }
@Override @Override
public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) { public boolean updateThreadPool(ThreadPoolAdapterParameter threadPoolAdapterParameter) {
String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey(); String threadPoolKey = threadPoolAdapterParameter.getThreadPoolKey();
ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(threadPoolKey); ThreadPoolExecutor threadPoolTaskExecutor = RABBITMQ_THREAD_POOL_TASK_EXECUTOR.get(threadPoolKey);
if (Objects.nonNull(threadPoolTaskExecutor)) { if (Objects.nonNull(threadPoolTaskExecutor)) {
int originalCoreSize = threadPoolTaskExecutor.getCorePoolSize(); int originalCoreSize = threadPoolTaskExecutor.getCorePoolSize();
int originalMaximumPoolSize = threadPoolTaskExecutor.getMaximumPoolSize(); int originalMaximumPoolSize = threadPoolTaskExecutor.getMaximumPoolSize();
threadPoolTaskExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize()); threadPoolTaskExecutor.setMaximumPoolSize(threadPoolAdapterParameter.getMaximumPoolSize());
threadPoolTaskExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize()); threadPoolTaskExecutor.setCorePoolSize(threadPoolAdapterParameter.getCorePoolSize());
log.info("[{}] Rabbitmq consumption thread pool parameter change. coreSize: {}, maximumSize: {}", log.info("[{}] Rabbitmq consumption thread pool parameter change. coreSize: {}, maximumSize: {}",
threadPoolKey, threadPoolKey,
String.format(CHANGE_DELIMITER, originalCoreSize, threadPoolAdapterParameter.getCorePoolSize()), String.format(CHANGE_DELIMITER, originalCoreSize, threadPoolAdapterParameter.getCorePoolSize()),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, threadPoolAdapterParameter.getMaximumPoolSize())); String.format(CHANGE_DELIMITER, originalMaximumPoolSize, threadPoolAdapterParameter.getMaximumPoolSize()));
return true; return true;
} }
log.warn("[{}] Rabbitmq consuming thread pool not found.", threadPoolKey); log.warn("[{}] Rabbitmq consuming thread pool not found.", threadPoolKey);
return false; return false;
} }
@Override @Override
public void onApplicationEvent(ApplicationStartedEvent event) { public void onApplicationEvent(ApplicationStartedEvent event) {
abstractConnectionFactoryMap.forEach((beanName, abstractConnectionFactor) -> { abstractConnectionFactoryMap.forEach((beanName, abstractConnectionFactor) -> {
ExecutorService executor = (ExecutorService) ReflectUtil.getFieldValue(abstractConnectionFactor, FiledName); ExecutorService executor = (ExecutorService) ReflectUtil.getFieldValue(abstractConnectionFactor, FiledName);
if (Objects.nonNull(executor)) { if (Objects.nonNull(executor)) {
if (executor instanceof ThreadPoolExecutor) { if (executor instanceof ThreadPoolExecutor) {
ThreadPoolExecutor threadPoolTaskExecutor = (ThreadPoolExecutor) executor; ThreadPoolExecutor threadPoolTaskExecutor = (ThreadPoolExecutor) executor;
RABBITMQ_THREAD_POOL_TASK_EXECUTOR.put(beanName, threadPoolTaskExecutor); RABBITMQ_THREAD_POOL_TASK_EXECUTOR.put(beanName, threadPoolTaskExecutor);
log.info("Rabbitmq executor name {}", beanName); log.info("Rabbitmq executor name {}", beanName);
} } else {
else { log.warn("Custom thread pools only support ThreadPoolExecutor");
log.warn("Custom thread pools only support ThreadPoolExecutor"); }
} }
}
});
}); }
}
} }

Write Preview
Loading…
Cancel
Save