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

Core code refactoring

Browse Source
pull/218/head
chen.ma 2 years ago
parent b281c150c1
commit 8d00d6e839
48 changed files with 74 additions and 252 deletions
Show Stats Download Patch File Download Diff File

22
hippo4j-common/src/main/java/cn/hippo4j/common/constant/ChangeThreadPoolConstants.java
Unescape View File

@ -0,0 +1,22 @@
package cn.hippo4j.common.constant;
/**
* Change thread-pool constants.
*
* @author chen.ma
* @date 2022/5/1 17:25
*/
public class ChangeThreadPoolConstants {
public static final String CHANGE_THREAD_POOL_TEXT = "[{}] Changed thread pool. " +
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]" +
"\n queueType :: [{}]" +
"\n capacity :: [{}]" +
"\n keepAliveTime :: [{}]" +
"\n executeTimeOut :: [{}]" +
"\n rejectedType :: [{}]" +
"\n allowCoreThreadTimeOut :: [{}]";
public static final String CHANGE_DELIMITER = "%s => %s";
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/config/BootstrapPropertiesInterface.java
Unescape View File

@ -87,5 +87,4 @@ public interface BootstrapPropertiesInterface {
default Boolean getBanner() {
return null;
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/config/ConfigEmptyException.java
Unescape View File

@ -39,5 +39,4 @@ public class ConfigEmptyException extends RuntimeException {
* action
*/
private String action;
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/config/UtilAutoConfiguration.java
Unescape View File

@ -37,5 +37,4 @@ public class UtilAutoConfiguration {
public InetUtils hippo4JInetUtils(InetUtilsProperties properties) {
return new InetUtils(properties);
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/config/WebThreadPoolConfiguration.java
Unescape View File

@ -77,5 +77,4 @@ public class WebThreadPoolConfiguration {
public WebThreadPoolHandlerChoose webThreadPoolServiceChoose() {
return new WebThreadPoolHandlerChoose();
}
}

6
hippo4j-core/src/main/java/cn/hippo4j/core/enable/BeforeCheckConfiguration.java
Unescape View File

@ -51,35 +51,30 @@ public class BeforeCheckConfiguration {
"Web server failed to start. The dynamic thread pool username is empty.",
"Please check whether the [spring.dynamic.thread-pool.username] configuration is empty or an empty string.");
}
String password = properties.getPassword();
if (StringUtil.isBlank(password)) {
throw new ConfigEmptyException(
"Web server failed to start. The dynamic thread pool password is empty.",
"Please check whether the [spring.dynamic.thread-pool.password] configuration is empty or an empty string.");
}
String namespace = properties.getNamespace();
if (StringUtil.isBlank(namespace)) {
throw new ConfigEmptyException(
"Web server failed to start. The dynamic thread pool namespace is empty.",
"Please check whether the [spring.dynamic.thread-pool.namespace] configuration is empty or an empty string.");
}
String itemId = properties.getItemId();
if (StringUtil.isBlank(itemId)) {
throw new ConfigEmptyException(
"Web server failed to start. The dynamic thread pool item id is empty.",
"Please check whether the [spring.dynamic.thread-pool.item-id] configuration is empty or an empty string.");
}
String serverAddr = properties.getServerAddr();
if (StringUtil.isBlank(serverAddr)) {
throw new ConfigEmptyException(
"Web server failed to start. The dynamic thread pool server addr is empty.",
"Please check whether the [spring.dynamic.thread-pool.server-addr] configuration is empty or an empty string.");
}
String applicationName = environment.getProperty("spring.application.name");
if (StringUtil.isBlank(applicationName)) {
throw new ConfigEmptyException(
@ -94,5 +89,4 @@ public class BeforeCheckConfiguration {
public class BeforeCheck {
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/enable/MarkerConfiguration.java
Unescape View File

@ -37,5 +37,4 @@ public class MarkerConfiguration {
public class Marker {
}
}

2
hippo4j-core/src/main/java/cn/hippo4j/core/executor/DynamicThreadPoolExecutor.java
Unescape View File

@ -71,11 +71,9 @@ public class DynamicThreadPoolExecutor extends AbstractDynamicExecutorSupport {
super(corePoolSize, maximumPoolSize, keepAliveTime, unit, waitForTasksToCompleteOnShutdown, awaitTerminationMillis, workQueue, threadPoolId, threadFactory, handler);
this.threadPoolId = threadPoolId;
this.executeTimeOut = executeTimeOut;
// Number of dynamic proxy denial policies.
RejectedExecutionHandler rejectedProxy = RejectedProxyUtil.createProxy(handler, threadPoolId, rejectCount);
setRejectedExecutionHandler(rejectedProxy);
// Redundant fields to avoid reflecting the acquired fields when sending change information.
redundancyHandler = handler;
}

61
hippo4j-core/src/main/java/cn/hippo4j/core/executor/DynamicThreadPoolWrapper.java
Unescape View File

@ -35,83 +35,29 @@ import java.util.concurrent.ThreadPoolExecutor;
@Data
public class DynamicThreadPoolWrapper implements DisposableBean {
/**
* Tenant id
*/
private String tenantId;
private String tenantId, itemId, threadPoolId;
/**
* Item id
*/
private String itemId;
private boolean subscribeFlag, initFlag;
/**
* Thread pool id
*/
private String tpId;
/**
* Subscribe flag
*/
private boolean subscribeFlag;
/**
* Init flag
*/
private boolean initFlag;
/**
* executor
* {@link DynamicThreadPoolExecutor}
*/
private ThreadPoolExecutor executor;
/**
* 首选服务端线程池, 为空使用默认线程池 {@link CommonDynamicThreadPool#getInstance(String)}
*
* @param threadPoolId
*/
public DynamicThreadPoolWrapper(String threadPoolId) {
this(threadPoolId, CommonDynamicThreadPool.getInstance(threadPoolId));
}
/**
* 首选服务端线程池, 为空使用 threadPoolExecutor.
*
* @param threadPoolId
* @param threadPoolExecutor
*/
public DynamicThreadPoolWrapper(String threadPoolId, ThreadPoolExecutor threadPoolExecutor) {
this.tpId = threadPoolId;
this.threadPoolId = threadPoolId;
this.executor = threadPoolExecutor;
}
/**
* 提交任务.
*
* @param command
*/
public void execute(Runnable command) {
executor.execute(command);
}
/**
* 提交任务.
*
* @param task
* @return
*/
public Future<?> submit(Runnable task) {
return executor.submit(task);
}
/**
* 提交任务.
*
* @param task
* @param <T>
* @return
*/
public <T> Future<T> submit(Callable<T> task) {
return executor.submit(task);
}
@ -122,5 +68,4 @@ public class DynamicThreadPoolWrapper implements DisposableBean {
((AbstractDynamicExecutorSupport) executor).destroy();
}
}
}

16
hippo4j-core/src/main/java/cn/hippo4j/core/executor/ThreadPoolNotifyAlarmHandler.java
Unescape View File

@ -107,10 +107,8 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
if (hippoSendMessageService == null) {
return;
}
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(threadPoolId);
BlockingQueue blockIngQueue = threadPoolExecutor.getQueue();
int queueSize = blockIngQueue.size();
int capacity = queueSize + blockIngQueue.remainingCapacity();
int divide = CalculateUtil.divide(queueSize, capacity);
@ -133,7 +131,6 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
int activeCount = threadPoolExecutor.getActiveCount();
int maximumPoolSize = threadPoolExecutor.getMaximumPoolSize();
int divide = CalculateUtil.divide(activeCount, maximumPoolSize);
ThreadPoolNotifyAlarm threadPoolNotifyAlarm = GlobalNotifyAlarmManage.get(threadPoolId);
boolean isSend = threadPoolNotifyAlarm.getIsAlarm()
&& divide > threadPoolNotifyAlarm.getActiveAlarm();
@ -154,7 +151,6 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
if (Objects.isNull(threadPoolNotifyAlarm) || !threadPoolNotifyAlarm.getIsAlarm()) {
return;
}
ThreadPoolExecutor threadPoolExecutor = GlobalThreadPoolManage.getExecutorService(threadPoolId).getExecutor();
checkPoolRejectedAlarm(threadPoolId, threadPoolExecutor);
}
@ -186,19 +182,16 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
if (Objects.isNull(threadPoolNotifyAlarm) || !threadPoolNotifyAlarm.getIsAlarm()) {
return;
}
if (threadPoolExecutor instanceof DynamicThreadPoolExecutor) {
try {
AlarmNotifyRequest alarmNotifyRequest = buildAlarmNotifyReq(threadPoolExecutor);
alarmNotifyRequest.setThreadPoolId(threadPoolId);
alarmNotifyRequest.setExecuteTime(executeTime);
alarmNotifyRequest.setExecuteTimeOut(executeTimeOut);
String executeTimeoutTrace = TraceContextUtil.getAndRemove();
if (StringUtil.isNotBlank(executeTimeoutTrace)) {
alarmNotifyRequest.setExecuteTimeoutTrace(executeTimeoutTrace);
}
Runnable task = () -> hippoSendMessageService.sendAlarmMessage(NotifyTypeEnum.TIMEOUT, alarmNotifyRequest);
EXECUTE_TIMEOUT_EXECUTOR.execute(task);
} catch (Throwable ex) {
@ -217,7 +210,6 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
String appName = StrUtil.isBlank(itemId) ? applicationName : itemId;
request.setAppName(appName);
request.setIdentify(IdentifyUtil.getIdentify());
hippoSendMessageService.sendChangeMessage(request);
}
@ -229,10 +221,8 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
*/
public AlarmNotifyRequest buildAlarmNotifyReq(ThreadPoolExecutor threadPoolExecutor) {
AlarmNotifyRequest request = new AlarmNotifyRequest();
String appName = StrUtil.isBlank(itemId) ? applicationName : itemId;
request.setAppName(appName);
// 核心线程数
int corePoolSize = threadPoolExecutor.getCorePoolSize();
// 最大线程数
@ -245,7 +235,6 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
int largestPoolSize = threadPoolExecutor.getLargestPoolSize();
// 线程池中执行任务总数量 (有锁)
long completedTaskCount = threadPoolExecutor.getCompletedTaskCount();
request.setActive(active.toUpperCase());
request.setIdentify(IdentifyUtil.getIdentify());
request.setCorePoolSize(corePoolSize);
@ -254,7 +243,6 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
request.setActiveCount(activeCount);
request.setLargestPoolSize(largestPoolSize);
request.setCompletedTaskCount(completedTaskCount);
BlockingQueue<Runnable> queue = threadPoolExecutor.getQueue();
// 队列元素个数
int queueSize = queue.size();
@ -268,18 +256,14 @@ public class ThreadPoolNotifyAlarmHandler implements Runnable, CommandLineRunner
request.setCapacity(queueCapacity);
request.setQueueSize(queueSize);
request.setRemainingCapacity(remainingCapacity);
RejectedExecutionHandler rejectedExecutionHandler = threadPoolExecutor instanceof DynamicThreadPoolExecutor
? ((DynamicThreadPoolExecutor) threadPoolExecutor).getRedundancyHandler()
: threadPoolExecutor.getRejectedExecutionHandler();
request.setRejectedExecutionHandlerName(rejectedExecutionHandler.getClass().getSimpleName());
long rejectCount = threadPoolExecutor instanceof DynamicThreadPoolExecutor
? ((DynamicThreadPoolExecutor) threadPoolExecutor).getRejectCountNum()
: -1L;
request.setRejectCountNum(rejectCount);
return request;
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/manage/GlobalNotifyAlarmManage.java
Unescape View File

@ -54,5 +54,4 @@ public class GlobalNotifyAlarmManage {
public static void put(String key, ThreadPoolNotifyAlarm val) {
NOTIFY_ALARM_MAP.put(key, val);
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/manage/GlobalThreadPoolManage.java
Unescape View File

@ -113,5 +113,4 @@ public class GlobalThreadPoolManage {
public static Integer getThreadPoolNum() {
return listThreadPoolId().size();
}
}

4
hippo4j-core/src/main/java/cn/hippo4j/core/executor/state/AbstractThreadPoolRuntime.java
Unescape View File

@ -83,7 +83,6 @@ public abstract class AbstractThreadPoolRuntime {
String currentLoad = CalculateUtil.divide(activeCount, maximumPoolSize) + "";
// 峰值负载
String peakLoad = CalculateUtil.divide(largestPoolSize, maximumPoolSize) + "";
BlockingQueue<Runnable> queue = pool.getQueue();
// 队列元素个数
int queueSize = queue.size();
@ -93,7 +92,6 @@ public abstract class AbstractThreadPoolRuntime {
int remainingCapacity = queue.remainingCapacity();
// 队列容量
int queueCapacity = queueSize + remainingCapacity;
stateInfo.setCoreSize(corePoolSize);
stateInfo.setTpId(threadPoolId);
stateInfo.setPoolSize(poolSize);
@ -107,7 +105,6 @@ public abstract class AbstractThreadPoolRuntime {
stateInfo.setQueueRemainingCapacity(remainingCapacity);
stateInfo.setLargestPoolSize(largestPoolSize);
stateInfo.setCompletedTaskCount(completedTaskCount);
long rejectCount =
pool instanceof DynamicThreadPoolExecutor ? ((DynamicThreadPoolExecutor) pool).getRejectCountNum() : -1L;
stateInfo.setRejectCount(rejectCount);
@ -115,5 +112,4 @@ public abstract class AbstractThreadPoolRuntime {
stateInfo.setTimestamp(System.currentTimeMillis());
return supplement(stateInfo);
}
}

9
hippo4j-core/src/main/java/cn/hippo4j/core/executor/state/ThreadPoolRunStateHandler.java
Unescape View File

@ -59,19 +59,15 @@ public class ThreadPoolRunStateHandler extends AbstractThreadPoolRuntime {
ByteConvertUtil.getPrintSize(runtimeInfo.getTotalMemory()),
" / 最大可用: ",
ByteConvertUtil.getPrintSize(runtimeInfo.getMaxMemory())).toString();
poolRunStateInfo.setCurrentLoad(poolRunStateInfo.getCurrentLoad() + "%");
poolRunStateInfo.setPeakLoad(poolRunStateInfo.getPeakLoad() + "%");
String ipAddress = hippo4JInetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
poolRunStateInfo.setHost(ipAddress);
poolRunStateInfo.setMemoryProportion(memoryProportion);
poolRunStateInfo.setFreeMemory(ByteConvertUtil.getPrintSize(runtimeInfo.getFreeMemory()));
String threadPoolId = poolRunStateInfo.getTpId();
DynamicThreadPoolWrapper executorService = GlobalThreadPoolManage.getExecutorService(threadPoolId);
ThreadPoolExecutor pool = executorService.getExecutor();
String rejectedName;
if (pool instanceof AbstractDynamicExecutorSupport) {
rejectedName = ((DynamicThreadPoolExecutor) pool).getRedundancyHandler().getClass().getSimpleName();
@ -79,17 +75,12 @@ public class ThreadPoolRunStateHandler extends AbstractThreadPoolRuntime {
rejectedName = pool.getRejectedExecutionHandler().getClass().getSimpleName();
}
poolRunStateInfo.setRejectedName(rejectedName);
ManyPoolRunStateInfo manyPoolRunStateInfo = BeanUtil.toBean(poolRunStateInfo, ManyPoolRunStateInfo.class);
manyPoolRunStateInfo.setIdentify(CLIENT_IDENTIFICATION_VALUE);
String active = environment.getProperty("spring.profiles.active", "UNKNOWN");
manyPoolRunStateInfo.setActive(active.toUpperCase());
String threadPoolState = ThreadPoolStatusHandler.getThreadPoolState(pool);
manyPoolRunStateInfo.setState(threadPoolState);
return manyPoolRunStateInfo;
}
}

5
hippo4j-core/src/main/java/cn/hippo4j/core/executor/state/ThreadPoolStatusHandler.java
Unescape View File

@ -53,14 +53,12 @@ public class ThreadPoolStatusHandler {
try {
Method runStateLessThan = ReflectUtil.getMethodByName(ThreadPoolExecutor.class, "runStateLessThan");
cn.hippo4j.common.toolkit.ReflectUtil.setAccessible(runStateLessThan);
AtomicInteger ctl = (AtomicInteger) ReflectUtil.getFieldValue(executor, "ctl");
int shutdown = (int) ReflectUtil.getFieldValue(executor, "SHUTDOWN");
boolean runStateLessThanBool = ReflectUtil.invoke(executor, runStateLessThan, ctl.get(), shutdown);
if (runStateLessThanBool) {
return RUNNING;
}
Method runStateAtLeast = ReflectUtil.getMethodByName(ThreadPoolExecutor.class, "runStateAtLeast");
cn.hippo4j.common.toolkit.ReflectUtil.setAccessible(runStateAtLeast);
int terminated = (int) ReflectUtil.getFieldValue(executor, "TERMINATED");
@ -68,12 +66,9 @@ public class ThreadPoolStatusHandler {
return resultStatus;
} catch (Exception ex) {
log.error("Failed to get thread pool status.", ex);
EXCEPTION_FLAG.set(Boolean.FALSE);
}
}
return "UNKNOWN";
}
}

3
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/AbstractBuildThreadPoolTemplate.java
Unescape View File

@ -77,7 +77,6 @@ public class AbstractBuildThreadPoolTemplate {
} catch (IllegalArgumentException ex) {
throw new IllegalArgumentException("Error creating thread pool parameter.", ex);
}
executorService.allowCoreThreadTimeOut(initParam.allowCoreThreadTimeOut);
return executorService;
}
@ -111,7 +110,6 @@ public class AbstractBuildThreadPoolTemplate {
} catch (IllegalArgumentException ex) {
throw new IllegalArgumentException("Error creating thread pool parameter.", ex);
}
taskQueue.setExecutor(fastThreadPoolExecutor);
fastThreadPoolExecutor.allowCoreThreadTimeOut(initParam.allowCoreThreadTimeOut);
return fastThreadPoolExecutor;
@ -142,7 +140,6 @@ public class AbstractBuildThreadPoolTemplate {
} catch (IllegalArgumentException ex) {
throw new IllegalArgumentException(String.format("Error creating thread pool parameter. threadPool id :: %s", initParam.getThreadPoolId()), ex);
}
dynamicThreadPoolExecutor.setTaskDecorator(initParam.getTaskDecorator());
dynamicThreadPoolExecutor.allowCoreThreadTimeOut(initParam.allowCoreThreadTimeOut);
return dynamicThreadPoolExecutor;

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/AbstractDynamicExecutorSupport.java
Unescape View File

@ -169,5 +169,4 @@ public abstract class AbstractDynamicExecutorSupport extends ThreadPoolExecutor
}
}
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/CommonDynamicThreadPool.java
Unescape View File

@ -45,5 +45,4 @@ public class CommonDynamicThreadPool {
.build();
return poolExecutor;
}
}

7
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/QueueTypeEnum.java
Unescape View File

@ -96,14 +96,12 @@ public enum QueueTypeEnum {
.filter(each -> Objects.equals(each.name, blockingQueueName))
.findFirst()
.orElse(null);
if (queueTypeEnum != null) {
blockingQueue = createBlockingQueue(queueTypeEnum.type, capacity);
if (Objects.equals(blockingQueue.getClass().getSimpleName(), blockingQueueName)) {
return blockingQueue;
}
}
Collection<CustomBlockingQueue> customBlockingQueues = DynamicThreadPoolServiceLoader
.getSingletonServiceInstances(CustomBlockingQueue.class);
blockingQueue = Optional.ofNullable(blockingQueue)
@ -120,7 +118,6 @@ public enum QueueTypeEnum {
return new LinkedBlockingQueue(temCapacity);
}));
return blockingQueue;
}
@ -148,7 +145,6 @@ public enum QueueTypeEnum {
} else if (Objects.equals(type, RESIZABLE_LINKED_BLOCKING_QUEUE.type)) {
blockingQueue = new ResizableCapacityLinkedBlockIngQueue(capacity);
}
Collection<CustomBlockingQueue> customBlockingQueues = DynamicThreadPoolServiceLoader
.getSingletonServiceInstances(CustomBlockingQueue.class);
blockingQueue = Optional.ofNullable(blockingQueue).orElseGet(() -> customBlockingQueues.stream()
@ -156,7 +152,6 @@ public enum QueueTypeEnum {
.map(each -> each.generateBlockingQueue())
.findFirst()
.orElse(new LinkedBlockingQueue(capacity)));
return blockingQueue;
}
@ -170,8 +165,6 @@ public enum QueueTypeEnum {
Optional<QueueTypeEnum> queueTypeEnum = Arrays.stream(QueueTypeEnum.values())
.filter(each -> each.type == type)
.findFirst();
return queueTypeEnum.map(each -> each.name).orElse("");
}
}

3
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/RejectedPolicies.java
Unescape View File

@ -52,7 +52,6 @@ public class RejectedPolicies {
executor.execute(r);
}
}
}
/**
@ -71,7 +70,5 @@ public class RejectedPolicies {
log.error("Adding Queue task to thread pool failed.", e);
}
}
}
}

7
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/RejectedTypeEnum.java
Unescape View File

@ -101,18 +101,15 @@ public enum RejectedTypeEnum {
.filter(each -> Objects.equals(each.name, name))
.findFirst()
.orElse(null);
if (rejectedTypeEnum != null) {
return rejectedTypeEnum.rejectedHandler;
}
Collection<CustomRejectedExecutionHandler> customRejectedExecutionHandlers = DynamicThreadPoolServiceLoader
.getSingletonServiceInstances(CustomRejectedExecutionHandler.class);
Optional<RejectedExecutionHandler> customRejected = customRejectedExecutionHandlers.stream()
.filter(each -> Objects.equals(name, each.getName()))
.map(each -> each.generateRejected())
.findFirst();
return customRejected.orElse(ABORT_POLICY.rejectedHandler);
}
@ -127,7 +124,6 @@ public enum RejectedTypeEnum {
.filter(each -> Objects.equals(type, each.type))
.map(each -> each.rejectedHandler)
.findFirst();
// 使用 SPI 匹配拒绝策略
RejectedExecutionHandler resultRejected = rejectedTypeEnum.orElseGet(() -> {
Collection<CustomRejectedExecutionHandler> customRejectedExecutionHandlers = DynamicThreadPoolServiceLoader
@ -136,10 +132,8 @@ public enum RejectedTypeEnum {
.filter(each -> Objects.equals(type, each.getType()))
.map(each -> each.generateRejected())
.findFirst();
return customRejected.orElse(ABORT_POLICY.rejectedHandler);
});
return resultRejected;
}
@ -152,5 +146,4 @@ public enum RejectedTypeEnum {
public static String getRejectedNameByType(int type) {
return createPolicy(type).getClass().getSimpleName();
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/ThreadFactoryBuilder.java
Unescape View File

@ -171,5 +171,4 @@ public class ThreadFactoryBuilder implements Builder<ThreadFactory> {
return thread;
};
}
}

8
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/ThreadPoolBuilder.java
Unescape View File

@ -173,6 +173,14 @@ public class ThreadPoolBuilder implements Builder<ThreadPoolExecutor> {
return this;
}
public ThreadPoolBuilder singlePool(String threadNamePrefix) {
int singleNum = 1;
this.corePoolSize = singleNum;
this.maxPoolSize = singleNum;
this.threadNamePrefix = threadNamePrefix;
return this;
}
public ThreadPoolBuilder poolThreadSize(int corePoolSize, int maxPoolSize) {
this.corePoolSize = corePoolSize;
this.maxPoolSize = maxPoolSize;

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/support/ThreadPoolExecutorTemplate.java
Unescape View File

@ -79,5 +79,4 @@ public class ThreadPoolExecutorTemplate extends ThreadPoolExecutor {
}
};
}
}

2
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/AbstractWebThreadPoolService.java
Unescape View File

@ -60,7 +60,6 @@ public abstract class AbstractWebThreadPoolService implements WebThreadPoolServi
}
}
}
return executor;
}
@ -72,5 +71,4 @@ public abstract class AbstractWebThreadPoolService implements WebThreadPoolServi
// ignore. Adaptation unit test.
}
}
}

16
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/JettyWebThreadPoolHandler.java
Unescape View File

@ -30,6 +30,8 @@ import org.springframework.boot.web.server.WebServer;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Executor;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
/**
* @author : wh
* @date : 2022/2/28 16:55
@ -50,10 +52,8 @@ public class JettyWebThreadPoolHandler extends AbstractWebThreadPoolService {
QueuedThreadPool queuedThreadPool = (QueuedThreadPool) executor;
poolBaseInfo.setCoreSize(queuedThreadPool.getMinThreads());
poolBaseInfo.setMaximumSize(queuedThreadPool.getMaxThreads());
BlockingQueue jobs = (BlockingQueue) ReflectUtil.getFieldValue(queuedThreadPool, "_jobs");
int queueCapacity = jobs.remainingCapacity() + jobs.size();
poolBaseInfo.setQueueCapacity(queueCapacity);
poolBaseInfo.setQueueType(jobs.getClass().getSimpleName());
poolBaseInfo.setKeepAliveTime((long) queuedThreadPool.getIdleTimeout());
@ -67,16 +67,13 @@ public class JettyWebThreadPoolHandler extends AbstractWebThreadPoolService {
try {
parameterInfo = new PoolParameterInfo();
QueuedThreadPool jettyExecutor = (QueuedThreadPool) executor;
int minThreads = jettyExecutor.getMinThreads();
int maxThreads = jettyExecutor.getMaxThreads();
parameterInfo.setCoreSize(minThreads);
parameterInfo.setMaxSize(maxThreads);
} catch (Exception ex) {
log.error("Failed to get the jetty thread pool parameter.", ex);
}
return parameterInfo;
}
@ -89,25 +86,20 @@ public class JettyWebThreadPoolHandler extends AbstractWebThreadPoolService {
public void updateWebThreadPool(PoolParameterInfo poolParameterInfo) {
try {
QueuedThreadPool jettyExecutor = (QueuedThreadPool) executor;
int minThreads = jettyExecutor.getMinThreads();
int maxThreads = jettyExecutor.getMaxThreads();
Integer coreSize = poolParameterInfo.getCoreSize();
Integer maxSize = poolParameterInfo.getMaxSize();
jettyExecutor.setMinThreads(coreSize);
jettyExecutor.setMaxThreads(maxSize);
log.info(
"[JETTY] Changed web thread pool. " +
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]",
String.format("%s => %s", minThreads, jettyExecutor.getMinThreads()),
String.format("%s => %s", maxThreads, jettyExecutor.getMaxThreads()));
String.format(CHANGE_DELIMITER, minThreads, jettyExecutor.getMinThreads()),
String.format(CHANGE_DELIMITER, maxThreads, jettyExecutor.getMaxThreads()));
} catch (Exception ex) {
log.error("Failed to modify the jetty thread pool parameter.", ex);
}
}
}

14
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/TomcatWebThreadPoolHandler.java
Unescape View File

@ -30,6 +30,8 @@ import org.springframework.boot.web.server.WebServer;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
/**
* Tomcat web thread pool handler.
*
@ -52,7 +54,6 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
log.warn("Exception getting Tomcat thread pool. Exception message :: {}", EXCEPTION_MESSAGE);
return null;
}
Executor tomcatExecutor = null;
try {
tomcatExecutor = ((TomcatWebServer) webServer).getTomcat().getConnector().getProtocolHandler().getExecutor();
@ -61,7 +62,6 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
EXCEPTION_MESSAGE = ex.getMessage();
log.error("Failed to get Tomcat thread pool. Message :: {}", EXCEPTION_MESSAGE);
}
return tomcatExecutor;
}
@ -85,7 +85,6 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
poolBaseInfo.setQueueType(queue.getClass().getSimpleName());
poolBaseInfo.setQueueCapacity(queueCapacity);
poolBaseInfo.setRejectedName(rejectedExecutionHandler.getClass().getSimpleName());
return poolBaseInfo;
}
@ -105,7 +104,6 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
} catch (Exception ex) {
log.error("Failed to get the tomcat thread pool parameter.", ex);
}
return parameterInfo;
}
@ -121,7 +119,6 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
int originalCoreSize = tomcatExecutor.getCorePoolSize();
int originalMaximumPoolSize = tomcatExecutor.getMaximumPoolSize();
long originalKeepAliveTime = tomcatExecutor.getKeepAliveTime(TimeUnit.SECONDS);
tomcatExecutor.setCorePoolSize(poolParameterInfo.getCoreSize());
tomcatExecutor.setMaximumPoolSize(poolParameterInfo.getMaxSize());
tomcatExecutor.setKeepAliveTime(poolParameterInfo.getKeepAliveTime(), TimeUnit.SECONDS);
@ -130,12 +127,11 @@ public class TomcatWebThreadPoolHandler extends AbstractWebThreadPoolService {
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]" +
"\n keepAliveTime :: [{}]",
String.format("%s => %s", originalCoreSize, poolParameterInfo.getCoreSize()),
String.format("%s => %s", originalMaximumPoolSize, poolParameterInfo.getMaxSize()),
String.format("%s => %s", originalKeepAliveTime, poolParameterInfo.getKeepAliveTime()));
String.format(CHANGE_DELIMITER, originalCoreSize, poolParameterInfo.getCoreSize()),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, poolParameterInfo.getMaxSize()),
String.format(CHANGE_DELIMITER, originalKeepAliveTime, poolParameterInfo.getKeepAliveTime()));
} catch (Exception ex) {
log.error("Failed to modify the Tomcat thread pool parameter.", ex);
}
}
}

16
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/UndertowWebThreadPoolHandler.java
Unescape View File

@ -38,6 +38,8 @@ import java.util.Date;
import java.util.Objects;
import java.util.concurrent.Executor;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
/**
* Undertow web thread pool handler.
*
@ -77,7 +79,6 @@ public class UndertowWebThreadPoolHandler extends AbstractWebThreadPoolService {
} catch (Exception ex) {
log.error("The undertow container failed to get thread pool parameters.", ex);
}
return poolBaseInfo;
}
@ -97,7 +98,6 @@ public class UndertowWebThreadPoolHandler extends AbstractWebThreadPoolService {
} catch (Exception ex) {
log.error("Failed to get the undertow thread pool parameter.", ex);
}
return parameterInfo;
}
@ -106,7 +106,6 @@ public class UndertowWebThreadPoolHandler extends AbstractWebThreadPoolService {
PoolRunStateInfo stateInfo = new PoolRunStateInfo();
XnioWorker xnioWorker = (XnioWorker) executor;
// private final TaskPool taskPool;
Field field = ReflectionUtils.findField(XnioWorker.class, "taskPool");
ReflectionUtils.makeAccessible(field);
Object fieldObject = ReflectionUtils.getField(field, xnioWorker);
@ -153,15 +152,12 @@ public class UndertowWebThreadPoolHandler extends AbstractWebThreadPoolService {
public void updateWebThreadPool(PoolParameterInfo poolParameterInfo) {
try {
XnioWorker xnioWorker = (XnioWorker) executor;
Integer coreSize = poolParameterInfo.getCoreSize();
Integer maxSize = poolParameterInfo.getMaxSize();
Integer keepAliveTime = poolParameterInfo.getKeepAliveTime();
int originalCoreSize = xnioWorker.getOption(Options.WORKER_TASK_CORE_THREADS);
int originalMaximumPoolSize = xnioWorker.getOption(Options.WORKER_TASK_MAX_THREADS);
int originalKeepAliveTime = xnioWorker.getOption(Options.WORKER_TASK_KEEPALIVE);
xnioWorker.setOption(Options.WORKER_TASK_CORE_THREADS, coreSize);
xnioWorker.setOption(Options.WORKER_TASK_MAX_THREADS, maxSize);
xnioWorker.setOption(Options.WORKER_TASK_KEEPALIVE, keepAliveTime);
@ -170,13 +166,11 @@ public class UndertowWebThreadPoolHandler extends AbstractWebThreadPoolService {
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]" +
"\n keepAliveTime :: [{}]",
String.format("%s => %s", originalCoreSize, coreSize),
String.format("%s => %s", originalMaximumPoolSize, maxSize),
String.format("%s => %s", originalKeepAliveTime, keepAliveTime));
String.format(CHANGE_DELIMITER, originalCoreSize, coreSize),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, maxSize),
String.format(CHANGE_DELIMITER, originalKeepAliveTime, keepAliveTime));
} catch (Exception ex) {
log.error("Failed to modify the undertow thread pool parameter.", ex);
}
}
}

2
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/WebThreadPoolHandlerChoose.java
Unescape View File

@ -42,8 +42,6 @@ public class WebThreadPoolHandlerChoose {
} catch (Exception ex) {
throw new ServiceException("Web thread pool service bean not found.", ex);
}
return webThreadPoolService;
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/WebThreadPoolRunStateHandler.java
Unescape View File

@ -51,5 +51,4 @@ public class WebThreadPoolRunStateHandler extends AbstractThreadPoolRuntime {
return poolRunStateInfo;
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/executor/web/WebThreadPoolService.java
Unescape View File

@ -66,5 +66,4 @@ public interface WebThreadPoolService {
* @param poolParameterInfo
*/
void updateWebThreadPool(PoolParameterInfo poolParameterInfo);
}

4
hippo4j-core/src/main/java/cn/hippo4j/core/handler/DynamicThreadPoolBannerHandler.java
Unescape View File

@ -58,16 +58,13 @@ public class DynamicThreadPoolBannerHandler implements InitializingBean {
"| |) | || | ' \\/ _` | ' \\| / _| | | | _/\n" +
"|___/ \\_, |_||_\\__,_|_|_|_|_\\__| |_| |_| \n" +
" |__/ \n";
if (properties.getBanner()) {
String version = getVersion();
version = (version != null) ? " (v" + version + ")" : "no version.";
StringBuilder padding = new StringBuilder();
while (padding.length() < STRAP_LINE_SIZE - (version.length() + DYNAMIC_THREAD_POOL.length())) {
padding.append(" ");
}
System.out.println(AnsiOutput.toString(banner, AnsiColor.GREEN, DYNAMIC_THREAD_POOL, AnsiColor.DEFAULT,
padding.toString(), AnsiStyle.FAINT, version, "\n\n", HIPPO4J_GITHUB, "\n", HIPPO4J_SITE, "\n"));
@ -78,5 +75,4 @@ public class DynamicThreadPoolBannerHandler implements InitializingBean {
final Package pkg = DynamicThreadPoolBannerHandler.class.getPackage();
return pkg != null ? pkg.getImplementationVersion() : "";
}
}

3
hippo4j-core/src/main/java/cn/hippo4j/core/proxy/RejectedProxyInvocationHandler.java
Unescape View File

@ -44,17 +44,14 @@ public class RejectedProxyInvocationHandler implements InvocationHandler {
@Override
public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
rejectCount.incrementAndGet();
if (ApplicationContextHolder.getInstance() != null) {
ThreadPoolNotifyAlarmHandler alarmHandler = ApplicationContextHolder.getBean(ThreadPoolNotifyAlarmHandler.class);
alarmHandler.checkPoolRejectedAlarm(threadPoolId);
}
try {
return method.invoke(target, args);
} catch (InvocationTargetException ex) {
throw ex.getCause();
}
}
}

2
hippo4j-core/src/main/java/cn/hippo4j/core/proxy/RejectedProxyUtil.java
Unescape View File

@ -43,8 +43,6 @@ public class RejectedProxyUtil {
rejectedExecutionHandler.getClass().getClassLoader(),
new Class[]{RejectedExecutionHandler.class},
new RejectedProxyInvocationHandler(rejectedExecutionHandler, threadPoolId, rejectedNum));
return rejectedProxy;
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/spi/CustomBlockingQueue.java
Unescape View File

@ -49,5 +49,4 @@ public interface CustomBlockingQueue {
* @return
*/
BlockingQueue generateBlockingQueue();
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/spi/CustomRejectedExecutionHandler.java
Unescape View File

@ -49,5 +49,4 @@ public interface CustomRejectedExecutionHandler {
* @return
*/
RejectedExecutionHandler generateRejected();
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/spi/DynamicThreadPoolServiceLoader.java
Unescape View File

@ -92,5 +92,4 @@ public class DynamicThreadPoolServiceLoader {
throw new ServiceLoaderInstantiationException(clazz, ex);
}
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/spi/ServiceLoaderInstantiationException.java
Unescape View File

@ -28,5 +28,4 @@ public class ServiceLoaderInstantiationException extends RuntimeException {
public ServiceLoaderInstantiationException(final Class<?> clazz, final Exception cause) {
super(String.format("Can not find public default constructor for SPI class `%s`", clazz.getName()), cause);
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/toolkit/CalculateUtil.java
Unescape View File

@ -35,5 +35,4 @@ public class CalculateUtil {
public static int divide(int num1, int num2) {
return ((int) (Double.parseDouble(num1 + "") / Double.parseDouble(num2 + "") * 100));
}
}

6
hippo4j-core/src/main/java/cn/hippo4j/core/toolkit/IdentifyUtil.java
Unescape View File

@ -61,7 +61,6 @@ public class IdentifyUtil {
port,
IDENTIFY_SLICER_SYMBOL,
CLIENT_IDENTIFICATION_VALUE).toString();
IDENTIFY = identification;
return identification;
}
@ -76,15 +75,12 @@ public class IdentifyUtil {
while (StrUtil.isBlank(IDENTIFY)) {
ConfigurableEnvironment environment = ApplicationContextHolder.getBean(ConfigurableEnvironment.class);
InetUtils inetUtils = ApplicationContextHolder.getBean(InetUtils.class);
if (environment != null && inetUtils != null) {
String identify = generate(environment, inetUtils);
return identify;
}
Thread.sleep(500);
}
return IDENTIFY;
}
@ -99,8 +95,6 @@ public class IdentifyUtil {
public static String getThreadPoolIdentify(String threadPoolId, String itemId, String namespace) {
ArrayList<String> params = Lists.newArrayList(
threadPoolId, itemId, namespace, getIdentify());
return Joiner.on(GROUP_KEY_DELIMITER).join(params);
}
}

1
hippo4j-core/src/main/java/cn/hippo4j/core/toolkit/TraceContextUtil.java
Unescape View File

@ -53,5 +53,4 @@ public class TraceContextUtil {
public static void setExecuteTimeoutTraceKey(String key) {
EXECUTE_TIMEOUT_TRACE_KEY = key;
}
}

37
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/AbstractCoreThreadPoolDynamicRefresh.java
Unescape View File

@ -54,6 +54,9 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_THREAD_POOL_TEXT;
/**
* Abstract core thread pool dynamic refresh.
*
@ -68,10 +71,7 @@ public abstract class AbstractCoreThreadPoolDynamicRefresh implements ThreadPool
protected final BootstrapCoreProperties bootstrapCoreProperties;
protected final ExecutorService dynamicRefreshExecutorService = ThreadPoolBuilder.builder()
.threadFactory("client.dynamic.refresh")
.singlePool()
.build();
protected final ExecutorService dynamicRefreshExecutorService = ThreadPoolBuilder.builder().singlePool("client.dynamic.refresh").build();
@Override
public void dynamicRefresh(String content) {
@ -176,26 +176,16 @@ public abstract class AbstractCoreThreadPoolDynamicRefresh implements ThreadPool
ExecutorProperties beforeProperties = GlobalCoreThreadPoolManage.getProperties(properties.getThreadPoolId());
// refresh executor properties
GlobalCoreThreadPoolManage.refresh(threadPoolId, properties);
log.info(
"[{}] Changed thread pool. " +
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]" +
"\n queueType :: [{}]" +
"\n capacity :: [{}]" +
"\n keepAliveTime :: [{}]" +
"\n executeTimeOut :: [{}]" +
"\n rejectedType :: [{}]" +
"\n allowCoreThreadTimeOut :: [{}]",
log.info(CHANGE_THREAD_POOL_TEXT,
threadPoolId.toUpperCase(),
String.format("%s => %s", beforeProperties.getCorePoolSize(), properties.getCorePoolSize()),
String.format("%s => %s", beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize()),
String.format("%s => %s", beforeProperties.getBlockingQueue(), properties.getBlockingQueue()),
String.format("%s => %s", beforeProperties.getQueueCapacity(), properties.getQueueCapacity()),
String.format("%s => %s", beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime()),
String.format("%s => %s", beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()),
String.format("%s => %s", beforeProperties.getRejectedHandler(), properties.getRejectedHandler()),
String.format("%s => %s", beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut()));
String.format(CHANGE_DELIMITER, beforeProperties.getCorePoolSize(), properties.getCorePoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize()),
String.format(CHANGE_DELIMITER, beforeProperties.getBlockingQueue(), properties.getBlockingQueue()),
String.format(CHANGE_DELIMITER, beforeProperties.getQueueCapacity(), properties.getQueueCapacity()),
String.format(CHANGE_DELIMITER, beforeProperties.getKeepAliveTime(), properties.getKeepAliveTime()),
String.format(CHANGE_DELIMITER, beforeProperties.getExecuteTimeOut(), properties.getExecuteTimeOut()),
String.format(CHANGE_DELIMITER, beforeProperties.getRejectedHandler(), properties.getRejectedHandler()),
String.format(CHANGE_DELIMITER, beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut()));
try {
threadPoolNotifyAlarmHandler.sendPoolConfigChange(newChangeRequest(beforeProperties, properties));
} catch (Throwable ex) {
@ -241,7 +231,6 @@ public abstract class AbstractCoreThreadPoolDynamicRefresh implements ThreadPool
private boolean checkConsistency(String threadPoolId, ExecutorProperties properties) {
ExecutorProperties beforeProperties = GlobalCoreThreadPoolManage.getProperties(properties.getThreadPoolId());
ThreadPoolExecutor executor = GlobalThreadPoolManage.getExecutorService(threadPoolId).getExecutor();
boolean result = !Objects.equals(beforeProperties.getCorePoolSize(), properties.getCorePoolSize())
|| !Objects.equals(beforeProperties.getMaximumPoolSize(), properties.getMaximumPoolSize())
|| !Objects.equals(beforeProperties.getAllowCoreThreadTimeOut(), properties.getAllowCoreThreadTimeOut())

1
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/ApolloRefresherHandler.java
Unescape View File

@ -62,5 +62,4 @@ public class ApolloRefresherHandler extends AbstractCoreThreadPoolDynamicRefresh
config.addChangeListener(configChangeListener);
log.info("dynamic-thread-pool refresher, add apollo listener success, namespace: {}", namespace);
}
}

18
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/BootstrapCorePropertiesBinderAdapt.java
Unescape View File

@ -66,7 +66,6 @@ public class BootstrapCorePropertiesBinderAdapt {
bindableCoreProperties = adapt(configInfo);
}
}
return bindableCoreProperties;
}
@ -86,14 +85,13 @@ public class BootstrapCorePropertiesBinderAdapt {
boolean containFlag = key != null
&& StringUtil.isNotBlank((String) key)
&& (((String) key).indexOf(PREFIX + ".executors") != -1
|| ((String) key).indexOf(PREFIX + ".notify-platforms") != -1
|| ((String) key).indexOf(PREFIX + ".notifyPlatforms") != -1);
|| ((String) key).indexOf(PREFIX + ".notify-platforms") != -1
|| ((String) key).indexOf(PREFIX + ".notifyPlatforms") != -1);
if (containFlag) {
String targetKey = key.toString().replace(PREFIX + ".", "");
targetMap.put(targetKey, val);
}
});
// convert
List<ExecutorProperties> executorPropertiesList = Lists.newArrayList();
List<NotifyPlatformProperties> notifyPropertiesList = Lists.newArrayList();
@ -111,7 +109,6 @@ public class BootstrapCorePropertiesBinderAdapt {
if (notifyKeySplit != null && notifyKeySplit.length > 0) {
key = key.replace("-", "_");
}
notifySingleMap.put(key, entry.getValue());
} else {
key = key.replace("executors[" + i + "].", "");
@ -123,27 +120,22 @@ public class BootstrapCorePropertiesBinderAdapt {
executorSingleMap.put(key, entry.getValue());
}
}
if (key.indexOf("notify-platforms[" + i + "].") != -1 || key.indexOf("notifyPlatforms[" + i + "].") != -1) {
if (key.indexOf("notify-platforms[" + i + "].") != -1) {
key = key.replace("notify-platforms[" + i + "].", "");
} else {
key = key.replace("notifyPlatforms[" + i + "].", "");
}
String[] keySplit = key.split("-");
if (keySplit != null && keySplit.length > 0) {
key = key.replace("-", "_");
}
platformSingleMap.put(key, entry.getValue());
}
}
if (CollectionUtil.isEmpty(executorSingleMap) && CollectionUtil.isEmpty(platformSingleMap)) {
break;
}
if (CollectionUtil.isNotEmpty(executorSingleMap)) {
ExecutorProperties executorProperties = BeanUtil.mapToBean(executorSingleMap, ExecutorProperties.class, true, CopyOptions.create());
if (executorProperties != null) {
@ -156,15 +148,12 @@ public class BootstrapCorePropertiesBinderAdapt {
notifyReceivesMap.put(value.name(), (String) receives);
}
}
alarm.setReceives(notifyReceivesMap);
executorProperties.setNotify(alarm);
}
executorPropertiesList.add(executorProperties);
}
}
if (CollectionUtil.isNotEmpty(platformSingleMap)) {
NotifyPlatformProperties notifyPlatformProperties = BeanUtil.mapToBean(platformSingleMap, NotifyPlatformProperties.class, true, CopyOptions.create());
if (notifyPlatformProperties != null) {
@ -172,15 +161,12 @@ public class BootstrapCorePropertiesBinderAdapt {
}
}
}
bindableCoreProperties = new BootstrapCoreProperties();
bindableCoreProperties.setExecutors(executorPropertiesList);
bindableCoreProperties.setNotifyPlatforms(notifyPropertiesList);
} catch (Exception ex) {
throw ex;
}
return bindableCoreProperties;
}
}

1
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/NacosCloudRefresherHandler.java
Unescape View File

@ -61,5 +61,4 @@ public class NacosCloudRefresherHandler extends AbstractCoreThreadPoolDynamicRef
}
});
}
}

1
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/NacosRefresherHandler.java
Unescape View File

@ -61,5 +61,4 @@ public class NacosRefresherHandler extends AbstractCoreThreadPoolDynamicRefresh
}
});
}
}

1
hippo4j-spring-boot/hippo4j-core-spring-boot-starter/src/main/java/cn/hippo4j/core/starter/refresher/ZookeeperRefresherHandler.java
Unescape View File

@ -112,5 +112,4 @@ public class ZookeeperRefresherHandler extends AbstractCoreThreadPoolDynamicRefr
log.error("load zk node error, nodePath is {}", nodePath, e);
}
}
}

30
hippo4j-spring-boot/hippo4j-spring-boot-starter/src/main/java/cn/hippo4j/starter/core/ServerThreadPoolDynamicRefresh.java
Unescape View File

@ -40,6 +40,9 @@ import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_DELIMITER;
import static cn.hippo4j.common.constant.ChangeThreadPoolConstants.CHANGE_THREAD_POOL_TEXT;
/**
* Thread pool dynamic refresh.
*
@ -110,26 +113,17 @@ public class ServerThreadPoolDynamicRefresh implements ThreadPoolDynamicRefresh
request.setNowExecuteTimeOut(originalExecuteTimeOut);
threadPoolNotifyAlarmHandler.sendPoolConfigChange(request);
log.info(
"[{}] Changed thread pool. " +
"\n coreSize :: [{}]" +
"\n maxSize :: [{}]" +
"\n queueType :: [{}]" +
"\n capacity :: [{}]" +
"\n keepAliveTime :: [{}]" +
"\n executeTimeOut :: [{}]" +
"\n rejectedType :: [{}]" +
"\n allowCoreThreadTimeOut :: [{}]",
log.info(CHANGE_THREAD_POOL_TEXT,
threadPoolId.toUpperCase(),
String.format("%s => %s", originalCoreSize, afterExecutor.getCorePoolSize()),
String.format("%s => %s", originalMaximumPoolSize, afterExecutor.getMaximumPoolSize()),
String.format("%s => %s", originalQuery, QueueTypeEnum.getBlockingQueueNameByType(parameter.getQueueType())),
String.format("%s => %s", originalCapacity,
String.format(CHANGE_DELIMITER, originalCoreSize, afterExecutor.getCorePoolSize()),
String.format(CHANGE_DELIMITER, originalMaximumPoolSize, afterExecutor.getMaximumPoolSize()),
String.format(CHANGE_DELIMITER, originalQuery, QueueTypeEnum.getBlockingQueueNameByType(parameter.getQueueType())),
String.format(CHANGE_DELIMITER, originalCapacity,
(afterExecutor.getQueue().remainingCapacity() + afterExecutor.getQueue().size())),
String.format("%s => %s", originalKeepAliveTime, afterExecutor.getKeepAliveTime(TimeUnit.SECONDS)),
String.format("%s => %s", originalExecuteTimeOut, originalExecuteTimeOut),
String.format("%s => %s", originalRejected, RejectedTypeEnum.getRejectedNameByType(parameter.getRejectedType())),
String.format("%s => %s", originalAllowCoreThreadTimeOut, EnableEnum.getBool(parameter.getAllowCoreThreadTimeOut())));
String.format(CHANGE_DELIMITER, originalKeepAliveTime, afterExecutor.getKeepAliveTime(TimeUnit.SECONDS)),
String.format(CHANGE_DELIMITER, originalExecuteTimeOut, originalExecuteTimeOut),
String.format(CHANGE_DELIMITER, originalRejected, RejectedTypeEnum.getRejectedNameByType(parameter.getRejectedType())),
String.format(CHANGE_DELIMITER, originalAllowCoreThreadTimeOut, EnableEnum.getBool(parameter.getAllowCoreThreadTimeOut())));
}
/**

Write Preview
Loading…
Cancel
Save