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

添加日志&清空ESå/redis的接口

Browse Source
This commit is contained in:
luke 2025-06-07 13:12:48 +08:00
parent 119c25060a
commit 9dda80b4dd
4 changed files with 168 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
src/main/java/com/knowledge/base/application/filter/AuthFilter.java
View File

@ -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)) {

37
src/main/java/com/knowledge/base/domain/doc/impl/AbstractBaseFileImporter.java
View File

@ -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<String> 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<Long> 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)) {

88
src/main/java/com/knowledge/base/infrastructure/north/controller/DocMaintenanceController.java Normal file
View File

@ -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<String> 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());
}
}
}

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

@ -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;
}
}