lukeye/doc-parser-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor: 优化线程池逻辑

Browse Source
This commit is contained in:
Luke.Ye 2025-08-27 14:46:03 +08:00
parent 8e5821a2d6
commit 705d403186
3 changed files with 164 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

47
src/main/java/com/knowledge/base/infrastructure/config/DynamicConfig.java
View File

@ -1,24 +1,30 @@
package com.knowledge.base.infrastructure.config;
import cn.hutool.core.util.StrUtil;
import com.knowledge.base.infrastructure.monitor.ThreadPoolMonitorStarter;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.context.config.annotation.RefreshScope;
import org.springframework.stereotype.Component;
import javax.annotation.PostConstruct;
import java.util.Arrays;
import java.util.Set;
import java.util.stream.Collectors;
/**
* @author Luke.ye
* @date 2025/5/8 09:59
* 动态配置类，支持 Nacos 热更新
*
* @author Luke
* @date 2025/5/8
*/
@Slf4j
@Getter
@Component
@RefreshScope
@Getter
public class DynamicConfig {
//是否记录请求和响应信息 Y-记录 N-不记录 默认记录
@Value("${micro.saas.doc.parser.recordMsgBody:Y}")
private String recordMsgBody;
@ -28,7 +34,6 @@ public class DynamicConfig {
@Value("${import.schedule.cron:* 0/30 * * * ?}")
private String importScheduleCron;
// token失效时间，默认24小时
@Value("${token.expire.time:86400000}")
private long tokenExpireTime;
@ -59,11 +64,41 @@ public class DynamicConfig {
@Value("${exclude.file.path.prefix}")
private String mdExcludePrefix;
@Value("${thread.pool.monitor.interval.seconds:60}")
private long threadPoolMonitorIntervalSeconds;
private boolean enableThreadPoolMonitor;
/**
* 来自 nacos 的开关控制是否启用线程池监控
*/
@Value("${thread.pool.monitor.enabled:true}")
public void setEnableThreadPoolMonitor(boolean enabled) {
boolean changed = this.enableThreadPoolMonitor != enabled;
this.enableThreadPoolMonitor = enabled;
if (changed) {
if (enabled) {
log.info("[config-refresh] enableThreadPoolMonitor=true，刷新线程池监控");
ThreadPoolMonitorStarter.getInstance().refreshAll();
} else {
log.info("[config-refresh] enableThreadPoolMonitor=false，清除线程池监控");
ThreadPoolMonitorStarter.getInstance().clearAll();
}
}
}
public Set<String> getLlmActiveSlugs() {
// 支持逗号、分号和换行分隔
return Arrays.stream(llmSharedActiveSlugIds.split("[,;\\n]"))
.map(String::trim)
.filter(StrUtil::isNotBlank)
.collect(Collectors.toSet());
}
@PostConstruct
public void init() {
log.info("[config-init] DynamicConfig 初始化完成，线程池监控开关 enableThreadPoolMonitor={}", enableThreadPoolMonitor);
if (enableThreadPoolMonitor) {
ThreadPoolMonitorStarter.getInstance().refreshAll();
}
}
}

87
src/main/java/com/knowledge/base/infrastructure/monitor/ThreadPoolMonitorStarter.java Normal file
View File

@ -0,0 +1,87 @@
package com.knowledge.base.infrastructure.monitor;
import cn.hutool.extra.spring.SpringUtil;
import com.knowledge.base.infrastructure.config.DynamicConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import javax.annotation.PreDestroy;
import java.util.Map;
import java.util.concurrent.*;
/**
* 线程池监控调度器（支持动态配置控制、自动刷新、清理）
*/
@Slf4j
@Component
@RequiredArgsConstructor
public class ThreadPoolMonitorStarter {
private final DynamicConfig dynamicConfig;
private final ScheduledExecutorService monitorScheduler = Executors.newSingleThreadScheduledExecutor(
r -> new Thread(r, "thread-monitor-scheduler"));
private final Map<String, ScheduledFuture<?>> monitorTasks = new ConcurrentHashMap<>();
private final Map<String, ExecutorService> registeredPools = new ConcurrentHashMap<>();
public void register(String poolName, ExecutorService pool) {
registeredPools.put(poolName, pool);
refresh(poolName, pool);
}
public void refreshAll() {
clearAll();
for (Map.Entry<String, ExecutorService> entry : registeredPools.entrySet()) {
refresh(entry.getKey(), entry.getValue());
}
}
public void refresh(String poolName, ExecutorService pool) {
if (!"Y".equalsIgnoreCase(dynamicConfig.getRecordMsgBody())) {
log.info("[thread-monitor] 未开启线程池监控配置，跳过 {}", poolName);
return;
}
if (!(pool instanceof ThreadPoolExecutor)) {
log.info("[thread-monitor] {} 非 ThreadPoolExecutor，无法监控", poolName);
return;
}
ThreadPoolExecutor executor = (ThreadPoolExecutor) pool;
ScheduledFuture<?> future = monitorScheduler.scheduleAtFixedRate(() -> {
log.info("[thread-monitor] {} - 活跃线程数: {}, 最大线程数: {}, 核心线程数: {}, 排队任务数: {}, 总任务数: {}, 已完成任务数: {}",
poolName,
executor.getActiveCount(),
executor.getMaximumPoolSize(),
executor.getCorePoolSize(),
executor.getQueue().size(),
executor.getTaskCount(),
executor.getCompletedTaskCount());
}, 0, 1, TimeUnit.MINUTES);
monitorTasks.put(poolName, future);
log.info("[thread-monitor] 线程池 {} 监控任务已启动", poolName);
}
public void clearAll() {
monitorTasks.forEach((name, future) -> {
future.cancel(true);
log.info("[thread-monitor] 已取消线程池 {} 的监控任务", name);
});
monitorTasks.clear();
}
@PreDestroy
public void destroy() {
clearAll();
monitorScheduler.shutdownNow();
}
public static ThreadPoolMonitorStarter getInstance() {
return SpringUtil.getBean(ThreadPoolMonitorStarter.class);
}
}

57
src/main/java/com/knowledge/base/infrastructure/util/ThreadPoolUtil.java
View File

@ -1,14 +1,16 @@
package com.knowledge.base.infrastructure.util;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import com.knowledge.base.infrastructure.monitor.ThreadPoolMonitorStarter;
import lombok.extern.slf4j.Slf4j;
import java.util.concurrent.*;
/**
* 通用线程池工具类
* 支持外部自定义线程池传入，未传入时使用默认线程池
* @author Luke
* 支持外部自定义线程池执行任务，并可选是否注册监控
*/
@Slf4j
public class ThreadPoolUtil {
private static final int CORE_POOL_SIZE = Runtime.getRuntime().availableProcessors() + 1;
@ -16,7 +18,6 @@ public class ThreadPoolUtil {
private static final int QUEUE_CAPACITY = 500;
private static final long KEEP_ALIVE_TIME = 60L;
// 默认线程池
private static final ThreadPoolExecutor DEFAULT_THREAD_POOL = new ThreadPoolExecutor(
CORE_POOL_SIZE,
MAX_POOL_SIZE,
@ -28,35 +29,52 @@ public class ThreadPoolUtil {
);
/**
* 执行任务，使用默认线程池
* 提交默认线程池任务
*/
public static void execute(Runnable task) {
DEFAULT_THREAD_POOL.execute(task);
DEFAULT_THREAD_POOL.execute(wrap(task, "default"));
}
/**
* 执行任务，允许调用方传入自定义线程池
* @param task Runnable
* @param executor 若为 null，则用默认线程池
* 提交自定义线程池任务（默认不监控）
*/
public static void execute(Runnable task, ExecutorService executor) {
if (executor != null) {
executor.execute(task);
} else {
DEFAULT_THREAD_POOL.execute(task);
execute(task, executor, false, "custom");
}
/**
* 提交任务（带监控选项 + 线程池名称）
*/
public static void execute(Runnable task, ExecutorService executor, boolean monitor, String poolName) {
if (executor == null) {
DEFAULT_THREAD_POOL.execute(wrap(task, "default"));
return;
}
executor.execute(wrap(task, poolName));
if (monitor) {
ThreadPoolMonitorStarter.getInstance().register(poolName, executor);
}
}
/**
* 优雅关闭（默认线程池）
*/
private static Runnable wrap(Runnable task, String poolName) {
return () -> {
String threadName = Thread.currentThread().getName();
try {
log.debug("[thread-pool][{}] 执行任务开始", poolName);
task.run();
log.debug("[thread-pool][{}] 执行任务结束", poolName);
} catch (Exception e) {
log.error("[thread-pool][{}] 执行异常", poolName, e);
} finally {
Thread.currentThread().setName(threadName); // 防止线程池复用导致名称混乱
}
};
}
public static void shutdownAndAwait() {
shutdownAndAwait(DEFAULT_THREAD_POOL);
}
/**
* 优雅关闭（指定线程池）
*/
public static void shutdownAndAwait(ExecutorService executor) {
if (executor == null) return;
executor.shutdown();
@ -70,9 +88,6 @@ public class ThreadPoolUtil {
}
}
/**
* 获取默认线程池（如需提交批量任务）
*/
public static ExecutorService getDefaultThreadPool() {
return DEFAULT_THREAD_POOL;
}