From 9dda80b4dd8642c19d04667ec73aebdd85f41b25 Mon Sep 17 00:00:00 2001 From: luke Date: Sat, 7 Jun 2025 13:12:48 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E6=97=A5=E5=BF=97&=E6=B8=85?= =?UTF-8?q?=E7=A9=BAES=C3=A5/redis=E7=9A=84=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../base/application/filter/AuthFilter.java | 4 +- .../doc/impl/AbstractBaseFileImporter.java | 37 +++++++- .../controller/DocMaintenanceController.java | 88 +++++++++++++++++++ .../infrastructure/util/ThreadPoolUtil.java | 50 +++++++++-- 4 files changed, 168 insertions(+), 11 deletions(-) create mode 100644 src/main/java/com/knowledge/base/infrastructure/north/controller/DocMaintenanceController.java diff --git a/src/main/java/com/knowledge/base/application/filter/AuthFilter.java b/src/main/java/com/knowledge/base/application/filter/AuthFilter.java index 3d9e8ea..2561844 100644 --- a/src/main/java/com/knowledge/base/application/filter/AuthFilter.java +++ b/src/main/java/com/knowledge/base/application/filter/AuthFilter.java @@ -33,7 +33,9 @@ public class AuthFilter implements Filter { String path = request.getRequestURI(); // 需要鉴权的路径前缀 - if (path.startsWith("/search")) { + if (path.startsWith("/search") + || path.startsWith("/api/v1/doc") + ) { String token = request.getHeader("Authorization"); if (StrUtil.isBlank(token)) { diff --git a/src/main/java/com/knowledge/base/domain/doc/impl/AbstractBaseFileImporter.java b/src/main/java/com/knowledge/base/domain/doc/impl/AbstractBaseFileImporter.java index af9cf55..a0456a6 100644 --- a/src/main/java/com/knowledge/base/domain/doc/impl/AbstractBaseFileImporter.java +++ b/src/main/java/com/knowledge/base/domain/doc/impl/AbstractBaseFileImporter.java @@ -20,12 +20,35 @@ import java.nio.file.*; import java.util.HashMap; import java.util.Map; import java.util.Optional; +import java.util.concurrent.*; public abstract class AbstractBaseFileImporter implements DocumentImporter { private static final Logger logger = LoggerFactory.getLogger(AbstractBaseFileImporter.class); private static final String INDEX_NAME = "documents"; + private static final ThreadPoolExecutor IMPORT_DOC_POOL = new ThreadPoolExecutor( + 8, + 8, + 60L, + TimeUnit.SECONDS, + new LinkedBlockingQueue<>(100), + new ThreadFactory() { + private final ThreadFactory defaultFactory = Executors.defaultThreadFactory(); + private int count = 1; + + @Override + public Thread newThread(Runnable r) { + Thread thread = defaultFactory.newThread(r); + thread.setName("Import-Doc-pool-" + count++); + thread.setDaemon(false); + return thread; + } + }, + new ThreadPoolExecutor.CallerRunsPolicy() + ); + + @Autowired protected RestHighLevelClient esClient; @@ -48,9 +71,16 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { Files.walk(basePath) .filter(p -> p.toString().toLowerCase().endsWith(getFileSuffix())) - .forEach(path -> ThreadPoolUtil.execute(() -> processFile(path, excludeBase))); + .forEach(path -> { + ThreadPoolUtil.execute(() -> processFile(path, excludeBase), IMPORT_DOC_POOL); + try { + Thread.sleep(100); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + }); - ThreadPoolUtil.shutdownAndAwait(); // 等待任务完成 + ThreadPoolUtil.shutdownAndAwait(); } private void processFile(Path path, Path excludeBase) { @@ -69,10 +99,10 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { Optional pathOpt = fileCacheService.getPath(fileName); if (pathOpt.isEmpty()) { fileCacheService.cachePath(fileName, relativePath, 24 * 60); + logger.info("新增relativePath: {} 文件，准备开始导入...... ", relativePath); } Long localMTime = Files.getLastModifiedTime(path).toMillis(); - logger.info("relativePath: {}， localMTime： {}", relativePath, localMTime); // 优先从缓存中加载数据，判断是否文件有修改 Optional lastModifiedTime = fileCacheService.getMTime(relativePath); if(lastModifiedTime.isPresent() && localMTime.equals(lastModifiedTime.get())) { @@ -86,6 +116,7 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter { return; } + logger.info("relativePath: {}， localMTime： {}", relativePath, localMTime); String docId = SafeIdUtil.encode(relativePath); GetRequest getRequest = new GetRequest(INDEX_NAME, docId); if (esClient.exists(getRequest, RequestOptions.DEFAULT)) { diff --git a/src/main/java/com/knowledge/base/infrastructure/north/controller/DocMaintenanceController.java b/src/main/java/com/knowledge/base/infrastructure/north/controller/DocMaintenanceController.java new file mode 100644 index 0000000..e04e23f --- /dev/null +++ b/src/main/java/com/knowledge/base/infrastructure/north/controller/DocMaintenanceController.java @@ -0,0 +1,88 @@ +package com.knowledge.base.infrastructure.north.controller; + +import com.knowledge.base.infrastructure.cache.FileCacheService; +import com.knowledge.base.infrastructure.util.SafeIdUtil; +import lombok.RequiredArgsConstructor; +import org.elasticsearch.action.delete.DeleteRequest; +import org.elasticsearch.client.RequestOptions; +import org.elasticsearch.client.RestHighLevelClient; +import org.elasticsearch.index.query.QueryBuilders; +import org.elasticsearch.index.reindex.DeleteByQueryRequest; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.springframework.http.ResponseEntity; +import org.springframework.web.bind.annotation.*; + +import java.util.Map; +import java.util.Optional; + +@RestController +@RequestMapping("/api/v1/doc") +@RequiredArgsConstructor +public class DocMaintenanceController { + + private final RestHighLevelClient esClient; + private final FileCacheService fileCacheService; + private static final String INDEX_NAME = "documents"; + private static final Logger logger = LoggerFactory.getLogger(DocMaintenanceController.class); + + /** + * 1. 根据文件名同时删除 ES 和缓存（Redis/本地）的记录 + */ + @DeleteMapping("/file/{fileName}") + public ResponseEntity deleteByFileName(@PathVariable String fileName) { + boolean esResult = false, cacheResult = false; + try { + // 1. 从缓存获取相对路径 + Optional relPathOpt = fileCacheService.getPath(fileName); + if (relPathOpt.isPresent()) { + String relativePath = relPathOpt.get(); + String docId = SafeIdUtil.encode(relativePath); + + // 2. 删除 ES + DeleteRequest deleteRequest = new DeleteRequest(INDEX_NAME, docId); + // 强制刷新可选：deleteRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE); + esClient.delete(deleteRequest, RequestOptions.DEFAULT); + esResult = true; + + // 3. 删除缓存（文件名和mtime都清除） + fileCacheService.clearPath(fileName); + fileCacheService.clearMTime(relativePath); + cacheResult = true; + + logger.info("已删除文件 [{}] 的ES和缓存记录", fileName); + } else { + logger.warn("未找到缓存记录: {}", fileName); + } + return ResponseEntity.ok(Map.of( + "esDeleted", esResult, + "cacheDeleted", cacheResult + )); + } catch (Exception e) { + logger.error("删除失败: {}", fileName, e); + return ResponseEntity.status(500).body("删除失败: " + e.getMessage()); + } + } + + /** + * 2. 同时清空 ES 和缓存中所有与文件相关的数据（危险操作！） + */ + @DeleteMapping("/files/clear-all") + public ResponseEntity clearAllFiles() { + try { + // 1. 清空缓存 + fileCacheService.clearAll(); + + // 2. 清空 ES documents 索引所有文档 + DeleteByQueryRequest deleteRequest = new DeleteByQueryRequest(INDEX_NAME); + deleteRequest.setQuery(QueryBuilders.matchAllQuery()); + esClient.deleteByQuery(deleteRequest, RequestOptions.DEFAULT); + + logger.warn("已清空 ES 和缓存中所有文件相关数据！"); + return ResponseEntity.ok(Map.of("msg", "所有文件相关数据已清空")); + } catch (Exception e) { + logger.error("全量清空失败", e); + return ResponseEntity.status(500).body("全量清空失败: " + e.getMessage()); + } + } +} diff --git a/src/main/java/com/knowledge/base/infrastructure/util/ThreadPoolUtil.java b/src/main/java/com/knowledge/base/infrastructure/util/ThreadPoolUtil.java index 30a2613..af1256b 100644 --- a/src/main/java/com/knowledge/base/infrastructure/util/ThreadPoolUtil.java +++ b/src/main/java/com/knowledge/base/infrastructure/util/ThreadPoolUtil.java @@ -4,6 +4,7 @@ import java.util.concurrent.*; /** * 通用线程池工具类 + * 支持外部自定义线程池传入，未传入时使用默认线程池 * @author Luke */ public class ThreadPoolUtil { @@ -13,7 +14,8 @@ public class ThreadPoolUtil { private static final int QUEUE_CAPACITY = 500; private static final long KEEP_ALIVE_TIME = 60L; - private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor( + // 默认线程池 + private static final ThreadPoolExecutor DEFAULT_THREAD_POOL = new ThreadPoolExecutor( CORE_POOL_SIZE, MAX_POOL_SIZE, KEEP_ALIVE_TIME, @@ -31,22 +33,56 @@ public class ThreadPoolUtil { return thread; } }, - new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略：主线程执行 + new ThreadPoolExecutor.CallerRunsPolicy() ); + /** + * 执行任务，使用默认线程池 + */ public static void execute(Runnable task) { - THREAD_POOL_EXECUTOR.execute(task); + DEFAULT_THREAD_POOL.execute(task); } + /** + * 执行任务，允许调用方传入自定义线程池 + * @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); + } + } + + /** + * 优雅关闭（默认线程池） + */ public static void shutdownAndAwait() { - THREAD_POOL_EXECUTOR.shutdown(); + shutdownAndAwait(DEFAULT_THREAD_POOL); + } + + /** + * 优雅关闭（指定线程池） + */ + public static void shutdownAndAwait(ExecutorService executor) { + if (executor == null) return; + executor.shutdown(); try { - if (!THREAD_POOL_EXECUTOR.awaitTermination(30, TimeUnit.MINUTES)) { - THREAD_POOL_EXECUTOR.shutdownNow(); + if (!executor.awaitTermination(30, TimeUnit.MINUTES)) { + executor.shutdownNow(); } } catch (InterruptedException e) { - THREAD_POOL_EXECUTOR.shutdownNow(); + executor.shutdownNow(); Thread.currentThread().interrupt(); } } + + /** + * 获取默认线程池（如需提交批量任务） + */ + public static ExecutorService getDefaultThreadPool() { + return DEFAULT_THREAD_POOL; + } }