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

上传文件至S3的同时，保存到本地

Browse Source
This commit is contained in:
luke 2025-06-18 18:41:24 +08:00
parent c1460acb64
commit 916e42bdf7
15 changed files with 206 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/main/java/com/knowledge/base/application/service/DocAppService.java
View File

@ -17,5 +17,5 @@ public interface DocAppService {
/**
* 保存上传记录
*/
void save(OSRecordDTO dto);
void saveOSUplodRecord(OSRecordDTO dto);
}

2
src/main/java/com/knowledge/base/application/service/DocAppServiceImpl.java
View File

@ -36,7 +36,7 @@ public class DocAppServiceImpl implements DocAppService{
}
@Override
public void save(OSRecordDTO dto) {
public void saveOSUplodRecord(OSRecordDTO dto) {
try {
fileDomainService.save(FileDtoConverter.toDO(dto));
} catch (Exception e) {

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

@ -2,6 +2,7 @@ package com.knowledge.base.domain.doc.service.impl;
import com.knowledge.base.domain.doc.service.iface.DocumentImporter;
import com.knowledge.base.infrastructure.cache.FileCacheService;
import com.knowledge.base.infrastructure.config.ThreadPoolConfig;
import com.knowledge.base.infrastructure.util.SafeIdUtil;
import com.knowledge.base.infrastructure.util.ThreadPoolUtil;
import org.elasticsearch.action.delete.DeleteRequest;
@ -27,27 +28,6 @@ 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;
@ -72,7 +52,7 @@ public abstract class AbstractBaseFileImporter implements DocumentImporter {
Files.walk(basePath)
.filter(p -> p.toString().toLowerCase().endsWith(getFileSuffix()))
.forEach(path -> {
ThreadPoolUtil.execute(() -> processFile(path, excludeBase), IMPORT_DOC_POOL);
ThreadPoolUtil.execute(() -> processFile(path, excludeBase), ThreadPoolConfig.IMPORT_DOC_POOL);
try {
Thread.sleep(100);
} catch (InterruptedException e) {

15
src/main/java/com/knowledge/base/infrastructure/cache/FileCacheService.java vendored
View File

@ -27,6 +27,21 @@ public interface FileCacheService {
*/
void cachePath(String filename, String relativePath, long expireMinutes);
/**
* 缓存文件名与文件相关的元信息
* @param filename 文件名，带后缀
* @param jsonMeta 文件元信息，包含：上传人、可访问链接、可访问链接过期时间、文件写入本地时间等
* @param expireMinutes 过期时间（分钟）
*/
default void cacheMeta(String filename, String jsonMeta, long expireMinutes) {};
/**
* 获取文件名对应的元信息
* @param filename
* @return
*/
default Optional<String> getMeta(String filename) { return null; };
/**
* 获取指定路径的上次修改时间（用于变更检测）
*

12
src/main/java/com/knowledge/base/infrastructure/cache/LocalFileCacheService.java vendored
View File

@ -50,6 +50,18 @@ public class LocalFileCacheService implements FileCacheService {
LocalCacheUtil.put(key("mtime", relativePath), mtime);
}
@Override
public void cacheMeta(String filename, String jsonMeta, long expireMinutes) {
LocalCacheUtil.put(key("meta", filename), jsonMeta);
}
@Override
public Optional<String> getMeta(String filename) {
Object val = LocalCacheUtil.get(key("meta", filename));
return val instanceof String ? Optional.of((String) val) : Optional.empty();
}
@Override
public void clearPath(String filename) {
LocalCacheUtil.remove(key("path", filename));

11
src/main/java/com/knowledge/base/infrastructure/cache/RedisFileCacheService.java vendored
View File

@ -42,6 +42,17 @@ public class RedisFileCacheService implements FileCacheService {
redisTemplate.opsForValue().set(key("path", filename), relativePath, expireMinutes, TimeUnit.MINUTES);
}
@Override
public void cacheMeta(String filename, String jsonMeta, long expireMinutes) {
redisTemplate.opsForValue().set(key("meta", filename), jsonMeta, expireMinutes, TimeUnit.MINUTES);
}
@Override
public Optional<String> getMeta(String filename) {
String value = redisTemplate.opsForValue().get(key("meta", filename));
return Optional.ofNullable(value);
}
@Override
public Optional<Long> getMTime(String relativePath) {
String val = redisTemplate.opsForValue().get(key("mtime", relativePath));

5
src/main/java/com/knowledge/base/infrastructure/config/ObjectStorageProperties.java
View File

@ -16,4 +16,9 @@ public class ObjectStorageProperties {
* 配置哪些 buckets 是公开的（不使用预签名链接）
*/
private List<String> publicBuckets;
/**
* 可被检索文件的本地路径
*/
private String localSearchablePath;
}

21
src/main/java/com/knowledge/base/infrastructure/config/ThreadPoolConfig.java Normal file
View File

@ -0,0 +1,21 @@
package com.knowledge.base.infrastructure.config;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
* @author Luke.ye
* @date 2025/6/18 14:32
*/
public class ThreadPoolConfig {
public static final ThreadPoolExecutor IMPORT_DOC_POOL = new ThreadPoolExecutor(
8,
8,
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(100),
ThreadFactoryBuilder.create().setNamePrefix("Import-Doc-pool-").build(),
new ThreadPoolExecutor.CallerRunsPolicy()
);
}

165
src/main/java/com/knowledge/base/infrastructure/north/controller/FileWriteController.java
View File

@ -1,7 +1,9 @@
package com.knowledge.base.infrastructure.north.controller;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.lang.UUID;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import com.knowledge.base.application.service.DocAppService;
import com.knowledge.base.application.service.UserAppService;
import com.knowledge.base.domain.doc.service.iface.DocumentImporter;
@ -12,6 +14,7 @@ import com.knowledge.base.infrastructure.north.dto.doc.OSRecordDTO;
import com.knowledge.base.infrastructure.north.dto.user.UserDTO;
import com.knowledge.base.infrastructure.north.dto.user.UserTokenDTO;
import com.knowledge.base.infrastructure.south.ObjectStorageGateway;
import com.knowledge.base.infrastructure.util.DateUtil;
import com.knowledge.base.infrastructure.util.SafeIdUtil;
import com.knowledge.base.infrastructure.util.ThreadPoolUtil;
import lombok.RequiredArgsConstructor;
@ -27,8 +30,16 @@ import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.multipart.MultipartFile;
import java.io.File;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardCopyOption;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
@RestController
@ -77,9 +88,6 @@ public class FileWriteController {
}
/**
* 1. 根据文件名同时删除 ES 和缓存（Redis/本地）的记录
*/
@DeleteMapping("/file/{fileName}")
public ResponseEntity<?> deleteByFileName(@PathVariable String fileName) {
boolean esResult = false, cacheResult = false;
@ -115,9 +123,6 @@ public class FileWriteController {
}
}
/**
* 2. 同时清空 ES 和缓存中所有与文件相关的数据（危险操作！）
*/
@DeleteMapping("/files/clear-all")
public ResponseEntity<?> clearAllFiles() {
try {
@ -141,77 +146,50 @@ public class FileWriteController {
public ResponseEntity<?> uploadFolderToOS(
@CookieValue(value = ConstantConfig.COOKIE_KEY, required = false) String cookieToken,
@RequestHeader(value = "Authorization", required = false) String token,
@RequestParam(value = "targetFolder", required = false, defaultValue = "default") String targetFolder,
@RequestParam(value = "searchable", required = false, defaultValue = "false") boolean searchable,
@RequestParam("files") MultipartFile[] files,
@RequestParam("bucket") String bucket
) {
try {
token = StrUtil.isBlank(token) ? cookieToken : token;
token = StrUtil.blankToDefault(token, cookieToken);
objectStorageGateway.ensureBucketExists(bucket);
Map<String, String> filePathMap = new LinkedHashMap<>();
for (MultipartFile file : files) {
// 获取原始文件名
String originalName = Optional.ofNullable(file.getOriginalFilename())
.filter(s -> !s.isBlank())
String originalName = Optional.ofNullable(file.getOriginalFilename()).filter(StrUtil::isNotBlank)
.orElse(file.getName());
// fileKey带类型后缀
String fileKey = Paths.get(originalName).getFileName().toString();
final String originFileNameWithSuffix = fileKey;
String suffix = FileUtil.getSuffix(fileKey).toLowerCase();
// 构造安全路径
String safeName = originalName.replace("\\", "/");
String folderPrefix = "default/";
// 构造路径
String folderPrefix = StrUtil.isBlank(targetFolder) ? "default/" : targetFolder;
if(!folderPrefix.endsWith("/")) {
folderPrefix += "/";
}
if (objectStorageProperties.isUseUuidPrefix()) {
String uuid = UUID.randomUUID().toString().replace("-", "").substring(0, 8);
safeName = folderPrefix + uuid + "_" + Paths.get(safeName).getFileName().toString();
} else {
safeName = folderPrefix + Paths.get(safeName).getFileName().toString();
fileKey = uuid + "_" + fileKey;
}
String s3FullPath = folderPrefix + fileKey;
// 上传文件到 MinIO
objectStorageGateway.uploadFile(bucket, safeName, file);
// 上传至对象存储
objectStorageGateway.uploadFile(bucket, s3FullPath, file);
String url = objectStorageGateway.generateUrl(bucket, s3FullPath);
filePathMap.put(originFileNameWithSuffix, url);
// 获取访问 URL
String fileKey = Paths.get(originalName).getFileName().toString();
String url = objectStorageGateway.generateUrl(bucket, safeName);
filePathMap.put(fileKey, url);
// 异步写入 OSS 上传记录
String finalSafeName = safeName;
// 异步处理上传记录与可检索文件
String finalToken = token;
ThreadPoolUtil.execute(() -> {
try {
Optional<UserTokenDTO> userTokenDTO = userAppService.findToken(finalToken);
if (userTokenDTO.isEmpty()) {
logger.warn("异步写入失败：无效 token");
return;
}
String finalUrl = url;
// file字段在主线程停止后，会自动删除，因此需要先转出来
byte[] fileBytes = file.getBytes();
Optional<UserDTO> userDTO = userAppService.findById(userTokenDTO.get().getUserId());
if (userDTO.isEmpty() || StrUtil.isBlank(userDTO.get().getUsername())) {
logger.warn("异步写入失败：无法获取用户名");
return;
}
LocalDateTime now = LocalDateTime.now();
boolean isPublic = objectStorageProperties.getPublicBuckets() != null &&
objectStorageProperties.getPublicBuckets().contains(bucket);
LocalDateTime expireTime = isPublic
? LocalDateTime.of(9999, 12, 31, 23, 59, 59)
: now.plusDays(7);
OSRecordDTO dto = new OSRecordDTO();
dto.setFileName(fileKey);
dto.setUrl(url);
dto.setUploader(userDTO.get().getUsername());
dto.setBucketName(bucket);
dto.setObjectPath(finalSafeName);
dto.setAddTime(now.toString());
dto.setUploadTime(now.toString());
dto.setExpireTime(expireTime.toString());
docAppService.save(dto);
} catch (Exception e) {
logger.error("异步写入 OSRecord 失败", e);
}
});
ThreadPoolUtil.execute(() ->
handleAsyncRecord(finalToken, bucket, originFileNameWithSuffix, suffix, s3FullPath, finalUrl, fileBytes, searchable)
);
}
return ResponseEntity.ok(Map.of(
@ -225,5 +203,72 @@ public class FileWriteController {
}
}
private void handleAsyncRecord(String token, String bucket, String originFileNameWithSuffix, String suffix,
String s3FullPath, String url, byte[] fileBytes, boolean searchable) {
try {
Optional<UserTokenDTO> userTokenDTO = userAppService.findToken(token);
if (userTokenDTO.isEmpty()) {
logger.warn("异步写入失败：无效 token");
return;
}
Optional<UserDTO> userDTO = userAppService.findById(userTokenDTO.get().getUserId());
if (userDTO.isEmpty() || StrUtil.isBlank(userDTO.get().getUsername())) {
logger.warn("异步写入失败：无法获取用户名");
return;
}
String uploader = userDTO.get().getUsername();
LocalDateTime now = LocalDateTime.now();
boolean isPublic = Optional.ofNullable(objectStorageProperties.getPublicBuckets())
.map(buckets -> buckets.contains(bucket))
.orElse(false);
LocalDateTime expireTime = isPublic
? LocalDateTime.of(9999, 12, 31, 23, 59, 59)
: now.plusDays(7);
OSRecordDTO dto = new OSRecordDTO();
dto.setFileName(originFileNameWithSuffix);
dto.setUrl(url);
dto.setUploader(uploader);
dto.setBucketName(bucket);
dto.setObjectPath(s3FullPath);
dto.setAddTime(now.toString());
dto.setUploadTime(now.toString());
dto.setExpireTime(expireTime.toString());
docAppService.saveOSUplodRecord(dto);
// 可检索文件的类型
Set<String> supportedTypes = Set.of("pdf", "doc", "docx", "xls", "xlsx");
if (searchable && supportedTypes.contains(suffix)) {
String dateFolder = now.toLocalDate().toString();
String localDir = Paths.get(objectStorageProperties.getLocalSearchablePath(), dateFolder).toString();
File localTargetDir = new File(localDir);
if (!localTargetDir.exists()) {
localTargetDir.mkdirs();
}
Path targetPath = Paths.get(localDir, originFileNameWithSuffix);
Files.write(targetPath, fileBytes);
long mtime = Files.getLastModifiedTime(targetPath).toMillis();
Map<String, Object> cacheMeta = new LinkedHashMap<>();
cacheMeta.put("fileName", originFileNameWithSuffix);
cacheMeta.put("uploader", uploader);
cacheMeta.put("url", url);
cacheMeta.put("bucket", bucket);
cacheMeta.put("expireTime", DateUtil.toMillis(expireTime));
cacheMeta.put("localAddTime", mtime);
fileCacheService.cacheMeta(originFileNameWithSuffix, JSONUtil.toJsonStr(cacheMeta), Duration.ofDays(2).toMinutes());
logger.info("本地文件信息已保存并写入缓存: {}", originFileNameWithSuffix);
}
} catch (Exception e) {
logger.error("异步写入失败: {}", originFileNameWithSuffix, e);
}
}
}

15
src/main/java/com/knowledge/base/infrastructure/util/DateUtil.java Normal file
View File

@ -0,0 +1,15 @@
package com.knowledge.base.infrastructure.util;
import java.time.LocalDateTime;
import java.time.ZoneId;
/**
* @author Luke.ye
* @date 2025/6/18 18:24
*/
public class DateUtil {
public static long toMillis(LocalDateTime time) {
return time.atZone(ZoneId.of("Asia/Shanghai")).toInstant().toEpochMilli();
}
}

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

@ -1,5 +1,7 @@
package com.knowledge.base.infrastructure.util;
import cn.hutool.core.thread.ThreadFactoryBuilder;
import java.util.concurrent.*;
/**
@ -21,18 +23,7 @@ public class ThreadPoolUtil {
KEEP_ALIVE_TIME,
TimeUnit.SECONDS,
new LinkedBlockingQueue<>(QUEUE_CAPACITY),
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("knowledge-base-default-pool-" + count++);
thread.setDaemon(false);
return thread;
}
},
ThreadFactoryBuilder.create().setNamePrefix("knowledge-base-default-pool-").build(),
new ThreadPoolExecutor.CallerRunsPolicy()
);

11
src/main/resources/application-dev-mac.properties
View File

@ -22,11 +22,12 @@ arthas.appName=${spring.application.name}
arthas.tunnel-server=ws://101.132.255.39:7777/ws
# 导入的材料路径
exclude.file.path.prefix=/Users/admin/Desktop/ahnx-share-src-public/public
markdown.path=/Users/admin/Desktop/ahnx-share-src-public/public
pdf.path=/Users/admin/Desktop/ahnx-share-src-public/public
word.path=/Users/admin/Desktop/ahnx-share-src-public/public
excel.path=/Users/admin/Desktop/ahnx-share-src-public/public
exclude.file.path.prefix=/Users/admin/Desktop/Archived/micro-saas
markdown.path=/Users/admin/Desktop/Archived/micro-saas
pdf.path=/Users/admin/Desktop/Archived/micro-saas
word.path=/Users/admin/Desktop/Archived/micro-saas
excel.path=/Users/admin/Desktop/Archived/micro-saas
object.storage.local-searchable-path=/Users/admin/Desktop/Archived/micro-saas
# mysql
spring.datasource.url=jdbc:mysql://localhost:3306/kbase?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai

7
src/main/resources/application-dev-windows.properties
View File

@ -24,9 +24,10 @@ arthas.tunnel-server=ws://101.132.255.39:7777/ws
# 导入的材料路径
exclude.file.path.prefix=D:/02-documents/01-ahnx-share-src-public/public
markdown.path=D:/02-documents/01-ahnx-share-src-public/public
pdf.path=D:/02-documents/01-ahnx-share-src-public/public
word.path=D:/02-documents/01-ahnx-share-src-public/public
excel.path=D:/02-documents/01-ahnx-share-src-public/public
pdf.path=D:/02-documents/01-os-uploaded-searchable-files
word.path=D:/02-documents/01-os-uploaded-searchable-files
excel.path=D:/02-documents/01-os-uploaded-searchable-files
object.storage.local-searchable-path=D:/02-documents/01-os-uploaded-searchable-file
# mysql

1
src/main/resources/application-docker.properties
View File

@ -26,6 +26,7 @@ markdown.path=/app/import-data
pdf.path=/app/import-data
word.path=/app/import-data
excel.path=/app/import-data
object.storage.local-searchable-path=/app/os-uploaded-searchable-files
# 相对路径裁剪
exclude.file.path.prefix=/app/import-data

7
src/main/resources/application.properties
View File

@ -3,6 +3,9 @@ spring.application.name=doc-parser-server
application.author=Luke.Ye
server.port=18080
spring.servlet.multipart.max-file-size=200MB
spring.servlet.multipart.max-request-size=500MB
# 日志相关
logging.config=classpath:log/log4j.xml
log4j2.enable.threadlocals=true
@ -22,12 +25,10 @@ import.schedule.enabled=true
# 每小时执行一次（可改）
import.schedule.cron=0 0 * * * *
# Redis 开关控制
knowledge.base.redis.enable=false
# 对象存储相关
object.storage.use-uuid-prefix=true
object.storage.public-buckets[0]=kbase
object.storage.public-buckets[1]=public-media
spring.servlet.multipart.max-file-size=200MB
spring.servlet.multipart.max-request-size=500MB