From 085408e84f61dcb294df157b22b5619318d3f009 Mon Sep 17 00:00:00 2001 From: dlwlrma <965810157@qq.com> Date: Tue, 29 Sep 2026 11:41:11 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=99=90=E5=88=B6=E7=9F=A5=E8=AF=86?= =?UTF-8?q?=E5=BA=93=E4=B8=8A=E4=BC=A0=E5=B9=B6=E5=8F=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 每实例默认最多四个同步上传，超限返回繁忙提示，正常或异常退出后释放名额 - 保留请求频率限制和原有上传链路，补充并发、异常释放与配置回归测试 - 最终后端 372 项通过、45 项原有禁用，前端及浏览器各 22 项通过；复验中的 TTS 超时记录保留 --- .../config/KnowledgeBaseUploadProperties.java | 18 ++ .../file/KnowledgeBaseUploadLimiter.java | 36 +++ .../service/KnowledgeBaseUploadService.java | 7 + app/src/main/resources/application.yml | 4 + .../KnowledgeBaseUploadPropertiesTest.java | 98 ++++++ .../file/KnowledgeBaseUploadLimiterTest.java | 35 +++ .../KnowledgeBaseUploadConcurrencyTest.java | 284 ++++++++++++++++++ .../KnowledgeBaseUploadServiceTest.java | 5 +- 8 files changed, 486 insertions(+), 1 deletion(-) create mode 100644 app/src/main/java/interview/guide/common/config/KnowledgeBaseUploadProperties.java create mode 100644 app/src/main/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiter.java create mode 100644 app/src/test/java/interview/guide/common/config/KnowledgeBaseUploadPropertiesTest.java create mode 100644 app/src/test/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiterTest.java create mode 100644 app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadConcurrencyTest.java diff --git a/app/src/main/java/interview/guide/common/config/KnowledgeBaseUploadProperties.java b/app/src/main/java/interview/guide/common/config/KnowledgeBaseUploadProperties.java new file mode 100644 index 00000000..2059dd04 --- /dev/null +++ b/app/src/main/java/interview/guide/common/config/KnowledgeBaseUploadProperties.java @@ -0,0 +1,18 @@ +package interview.guide.common.config; + +import jakarta.validation.constraints.Min; +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.stereotype.Component; +import org.springframework.validation.annotation.Validated; + +/** 知识库同步上传链路的单实例并发配置，修改后需重启应用。 */ +@Data +@Component +@Validated +@ConfigurationProperties(prefix = "app.knowledge-base-upload") +public class KnowledgeBaseUploadProperties { + + @Min(1) + private int maxConcurrent = 4; +} diff --git a/app/src/main/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiter.java b/app/src/main/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiter.java new file mode 100644 index 00000000..57365d47 --- /dev/null +++ b/app/src/main/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiter.java @@ -0,0 +1,36 @@ +package interview.guide.infrastructure.file; + +import interview.guide.common.config.KnowledgeBaseUploadProperties; +import interview.guide.common.exception.BusinessException; +import interview.guide.common.exception.ErrorCode; +import java.util.concurrent.Semaphore; +import java.util.function.Supplier; +import org.springframework.stereotype.Component; +import org.springframework.util.Assert; + +/** + * 限制一个应用实例中正在执行的知识库上传，不排队等待。 + * + *

不替代请求频率限制，也不限制 multipart 接收或后台向量化任务。 + */ +@Component +public class KnowledgeBaseUploadLimiter { + + private final Semaphore permits; + + public KnowledgeBaseUploadLimiter(KnowledgeBaseUploadProperties properties) { + Assert.isTrue(properties.getMaxConcurrent() > 0, "知识库上传并发上限必须大于 0"); + permits = new Semaphore(properties.getMaxConcurrent()); + } + + public T execute(Supplier upload) { + if (!permits.tryAcquire()) { + throw new BusinessException(ErrorCode.RATE_LIMIT_EXCEEDED, "知识库上传繁忙，请稍后重试"); + } + try { + return upload.get(); + } finally { + permits.release(); + } + } +} diff --git a/app/src/main/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadService.java b/app/src/main/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadService.java index d99b8152..f77071e9 100644 --- a/app/src/main/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadService.java +++ b/app/src/main/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadService.java @@ -5,6 +5,7 @@ import interview.guide.infrastructure.file.FileHashService; import interview.guide.infrastructure.file.FileStorageService; import interview.guide.infrastructure.file.FileValidationService; +import interview.guide.infrastructure.file.KnowledgeBaseUploadLimiter; import interview.guide.modules.knowledgebase.listener.VectorizeStreamProducer; import interview.guide.modules.knowledgebase.model.KnowledgeBaseEntity; import interview.guide.modules.knowledgebase.model.VectorStatus; @@ -34,6 +35,7 @@ public class KnowledgeBaseUploadService { private final FileValidationService fileValidationService; private final FileHashService fileHashService; private final VectorizeStreamProducer vectorizeStreamProducer; + private final KnowledgeBaseUploadLimiter uploadLimiter; private static final long MAX_FILE_SIZE = 50 * 1024 * 1024; // 50MB @@ -46,6 +48,11 @@ public class KnowledgeBaseUploadService { * @return 上传结果和存储信息（包含duplicate字段，表示是否为重复上传） */ public Map uploadKnowledgeBase(MultipartFile file, String name, String category) { + return uploadLimiter.execute(() -> doUploadKnowledgeBase(file, name, category)); + } + + private Map doUploadKnowledgeBase( + MultipartFile file, String name, String category) { // 1. 验证文件 fileValidationService.validateFile(file, MAX_FILE_SIZE, "知识库"); diff --git a/app/src/main/resources/application.yml b/app/src/main/resources/application.yml index 5b4bc56b..67e947fd 100644 --- a/app/src/main/resources/application.yml +++ b/app/src/main/resources/application.yml @@ -206,6 +206,10 @@ app: - application/vnd.openxmlformats-officedocument.wordprocessingml.document - text/plain + # 单个应用实例中同步处理的知识库上传数，不替代请求频率限制 + knowledge-base-upload: + max-concurrent: ${APP_KNOWLEDGE_BASE_UPLOAD_MAX_CONCURRENT:4} + # RustFS (S3兼容) 存储配置 storage: endpoint: ${APP_STORAGE_ENDPOINT:http://localhost:9000} diff --git a/app/src/test/java/interview/guide/common/config/KnowledgeBaseUploadPropertiesTest.java b/app/src/test/java/interview/guide/common/config/KnowledgeBaseUploadPropertiesTest.java new file mode 100644 index 00000000..ff9cba3f --- /dev/null +++ b/app/src/test/java/interview/guide/common/config/KnowledgeBaseUploadPropertiesTest.java @@ -0,0 +1,98 @@ +package interview.guide.common.config; + +import interview.guide.common.exception.BusinessException; +import interview.guide.infrastructure.file.KnowledgeBaseUploadLimiter; +import java.io.IOException; +import java.util.Map; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.springframework.boot.context.properties.EnableConfigurationProperties; +import org.springframework.boot.env.YamlPropertySourceLoader; +import org.springframework.boot.context.properties.bind.validation.BindValidationException; +import org.springframework.boot.test.context.runner.ApplicationContextRunner; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.core.env.MapPropertySource; +import org.springframework.core.io.ClassPathResource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +@DisplayName("上传并发配置绑定与校验") +class KnowledgeBaseUploadPropertiesTest { + + private final ApplicationContextRunner contextRunner = new ApplicationContextRunner() + .withUserConfiguration(UploadConfiguration.class); + + @Test + @DisplayName("未设置配置时应用使用默认四个上传名额") + void shouldUseDefaultLimit() { + contextRunner.run(context -> { + assertThat(context).hasNotFailed(); + assertThat(context.getBean(KnowledgeBaseUploadProperties.class).getMaxConcurrent()) + .isEqualTo(4); + assertThat(context.getBean(KnowledgeBaseUploadLimiter.class).execute(() -> "uploaded")) + .isEqualTo("uploaded"); + }); + } + + @Test + @DisplayName("自定义并发配置实际作用于共享上传名额") + void shouldBindConfiguredLimitToLimiter() { + contextRunner.withPropertyValues("app.knowledge-base-upload.max-concurrent=2").run(context -> { + assertThat(context).hasNotFailed(); + KnowledgeBaseUploadLimiter limiter = context.getBean(KnowledgeBaseUploadLimiter.class); + + limiter.execute(() -> limiter.execute(() -> { + assertThatThrownBy(() -> limiter.execute(() -> "excess upload")) + .isInstanceOf(BusinessException.class); + return "uploaded"; + })); + + assertThat(limiter.execute(() -> "uploaded")).isEqualTo("uploaded"); + }); + } + + @Test + @DisplayName("application.yml 中的环境变量占位符可覆盖默认上传上限") + void shouldApplyEnvironmentOverrideFromApplicationYaml() throws IOException { + var sources = new YamlPropertySourceLoader() + .load("application", new ClassPathResource("application.yml")); + contextRunner.withInitializer(context -> { + context.getEnvironment().getPropertySources().addFirst(new MapPropertySource( + "upload-override", Map.of("APP_KNOWLEDGE_BASE_UPLOAD_MAX_CONCURRENT", "1"))); + sources.forEach(source -> context.getEnvironment().getPropertySources().addLast(source)); + }).run(context -> { + assertThat(context).hasNotFailed(); + KnowledgeBaseUploadLimiter limiter = context.getBean(KnowledgeBaseUploadLimiter.class); + + limiter.execute(() -> { + assertThatThrownBy(() -> limiter.execute(() -> "excess upload")) + .isInstanceOf(BusinessException.class); + return "uploaded"; + }); + + assertThat(limiter.execute(() -> "uploaded")).isEqualTo("uploaded"); + }); + } + + @ParameterizedTest + @ValueSource(ints = {0, -1}) + @DisplayName("零或负数并发配置在启动时明确失败") + void shouldRejectInvalidLimit(int maximum) { + contextRunner.withPropertyValues("app.knowledge-base-upload.max-concurrent=" + maximum) + .run(context -> { + assertThat(context).hasFailed(); + assertThat(context.getStartupFailure()) + .hasRootCauseInstanceOf(BindValidationException.class); + }); + } + + @Configuration(proxyBeanMethods = false) + @EnableConfigurationProperties(KnowledgeBaseUploadProperties.class) + @Import(KnowledgeBaseUploadLimiter.class) + static class UploadConfiguration { + } +} diff --git a/app/src/test/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiterTest.java b/app/src/test/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiterTest.java new file mode 100644 index 00000000..7dd6ab17 --- /dev/null +++ b/app/src/test/java/interview/guide/infrastructure/file/KnowledgeBaseUploadLimiterTest.java @@ -0,0 +1,35 @@ +package interview.guide.infrastructure.file; + +import interview.guide.common.config.KnowledgeBaseUploadProperties; +import interview.guide.common.exception.BusinessException; +import interview.guide.common.exception.ErrorCode; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +@DisplayName("上传名额释放边界") +class KnowledgeBaseUploadLimiterTest { + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @DisplayName("业务异常与 Error 原样抛出，之后仍可再次执行上传") + void shouldReleasePermitAndPreserveFailure(boolean error) { + KnowledgeBaseUploadProperties properties = new KnowledgeBaseUploadProperties(); + properties.setMaxConcurrent(1); + KnowledgeBaseUploadLimiter limiter = new KnowledgeBaseUploadLimiter(properties); + BusinessException businessFailure = new BusinessException(ErrorCode.STORAGE_UPLOAD_FAILED); + AssertionError unexpectedFailure = new AssertionError("unexpected failure"); + + assertThatThrownBy(() -> limiter.execute(() -> { + if (error) { + throw unexpectedFailure; + } + throw businessFailure; + })).isSameAs(error ? unexpectedFailure : businessFailure); + + assertThat(limiter.execute(() -> "uploaded")).isEqualTo("uploaded"); + } +} diff --git a/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadConcurrencyTest.java b/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadConcurrencyTest.java new file mode 100644 index 00000000..cac521b2 --- /dev/null +++ b/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadConcurrencyTest.java @@ -0,0 +1,284 @@ +package interview.guide.modules.knowledgebase.service; + +import interview.guide.common.config.KnowledgeBaseUploadProperties; +import interview.guide.common.exception.BusinessException; +import interview.guide.common.exception.ErrorCode; +import interview.guide.common.exception.GlobalExceptionHandler; +import interview.guide.infrastructure.file.FileHashService; +import interview.guide.infrastructure.file.FileStorageService; +import interview.guide.infrastructure.file.FileValidationService; +import interview.guide.infrastructure.file.KnowledgeBaseUploadLimiter; +import interview.guide.modules.knowledgebase.KnowledgeBaseController; +import interview.guide.modules.knowledgebase.listener.VectorizeStreamProducer; +import interview.guide.modules.knowledgebase.model.KnowledgeBaseEntity; +import interview.guide.modules.knowledgebase.repository.KnowledgeBaseRepository; +import java.util.ArrayList; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.concurrent.ArrayBlockingQueue; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.mock.web.MockMultipartFile; +import org.springframework.web.multipart.MultipartFile; +import org.springframework.test.web.servlet.MockMvc; +import org.springframework.test.web.servlet.setup.MockMvcBuilders; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; +import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.multipart; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.jsonPath; +import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status; + +@ExtendWith(MockitoExtension.class) +@DisplayName("知识库上传并发保护") +class KnowledgeBaseUploadConcurrencyTest { + + @Mock + private KnowledgeBaseParseService parseService; + @Mock + private KnowledgeBasePersistenceService persistenceService; + @Mock + private FileStorageService storageService; + @Mock + private KnowledgeBaseRepository repository; + @Mock + private FileValidationService validationService; + @Mock + private FileHashService hashService; + @Mock + private VectorizeStreamProducer producer; + + private KnowledgeBaseUploadService service; + private MultipartFile file; + + @BeforeEach + void setUp() { + service = createService(4); + file = new MockMultipartFile("file", "guide.pdf", "application/pdf", new byte[] {1, 2, 3}); + } + + @ParameterizedTest + @ValueSource(booleans = {false, true}) + @DisplayName("四个上传处理中时拒绝第五个请求，完成后允许再次上传") + void shouldRejectFifthInFlightUploadAndAllowRetryAfterCompletion(boolean virtualThreads) + throws Exception { + CountDownLatch entered = new CountDownLatch(4); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger validationCalls = new AtomicInteger(); + doAnswer(invocation -> { + if (validationCalls.incrementAndGet() <= 4) { + entered.countDown(); + assertThat(release.await(10, TimeUnit.SECONDS)).isTrue(); + } + return null; + }).when(validationService).validateFile(any(), anyLong(), eq("知识库")); + stubDuplicateUpload(); + ThreadPoolExecutor executor = new ThreadPoolExecutor(4, 4, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(4), virtualThreads + ? Thread.ofVirtual().name("upload-regression-", 0).factory() + : Thread.ofPlatform().name("upload-regression-", 0).factory()); + List>> uploads = new ArrayList<>(); + + try { + for (int index = 0; index < 4; index++) { + uploads.add(executor.submit(() -> service.uploadKnowledgeBase(file, null, null))); + } + assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue(); + + assertThatThrownBy(() -> service.uploadKnowledgeBase(file, null, null)) + .isInstanceOfSatisfying(BusinessException.class, exception -> { + assertThat(exception.getCode()).isEqualTo(ErrorCode.RATE_LIMIT_EXCEEDED.getCode()); + assertThat(exception.getMessage()).contains("上传繁忙", "重试"); + }); + assertThatThrownBy(() -> service.uploadKnowledgeBase(file, null, null)) + .isInstanceOf(BusinessException.class); + + assertThat(validationCalls).hasValue(4); + release.countDown(); + for (Future> upload : uploads) { + assertThat(upload.get(5, TimeUnit.SECONDS)).containsEntry("duplicate", true); + } + assertThat(service.uploadKnowledgeBase(file, null, null)).containsEntry("duplicate", true); + verify(storageService, never()).uploadKnowledgeBase(any()); + verify(producer, never()).sendVectorizeTask(anyLong()); + } finally { + release.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + @DisplayName("校验异常不会泄漏上传名额，原始业务异常保持不变") + void shouldReleasePermitAfterValidationFailure() { + service = createService(1); + BusinessException failure = new BusinessException(ErrorCode.BAD_REQUEST, "文件校验失败"); + doThrow(failure).doNothing().when(validationService) + .validateFile(any(), anyLong(), eq("知识库")); + stubDuplicateUpload(); + + assertThatThrownBy(() -> service.uploadKnowledgeBase(file, null, null)).isSameAs(failure); + + assertThat(service.uploadKnowledgeBase(file, null, null)).containsEntry("duplicate", true); + } + + @Test + @DisplayName("重复上传处理异常也会释放名额，并保留原始异常") + void shouldReleasePermitAfterDuplicateHandlingFailure() { + service = createService(1); + KnowledgeBaseEntity existing = stubDuplicateUpload(); + BusinessException failure = new BusinessException(ErrorCode.INTERNAL_ERROR, "数据库读取失败"); + when(persistenceService.handleDuplicateKnowledgeBase(existing, "existing-hash")) + .thenThrow(failure).thenReturn(Map.of("duplicate", true)); + + assertThatThrownBy(() -> service.uploadKnowledgeBase(file, null, null)).isSameAs(failure); + + assertThat(service.uploadKnowledgeBase(file, null, null)).containsEntry("duplicate", true); + } + + @ParameterizedTest + @ValueSource(booleans = {true, false}) + @DisplayName("存储或数据库保存失败后释放名额，保留原异常和孤儿对象补偿") + void shouldReleasePermitAfterStorageOrDatabaseFailure(boolean storageFails) { + service = createService(1); + KnowledgeBaseEntity existing = stubDuplicateUpload(); + when(repository.findByFileHash("existing-hash")) + .thenReturn(Optional.empty()).thenReturn(Optional.of(existing)); + BusinessException failure = new BusinessException( + storageFails ? ErrorCode.STORAGE_UPLOAD_FAILED : ErrorCode.INTERNAL_ERROR, "上传依赖失败"); + if (storageFails) { + when(storageService.uploadKnowledgeBase(file)).thenThrow(failure); + } else { + when(storageService.uploadKnowledgeBase(file)).thenReturn("kb/orphan"); + when(storageService.getFileUrl("kb/orphan")).thenReturn("http://storage/kb/orphan"); + when(persistenceService.saveKnowledgeBase(any(), any(), any(), anyString(), anyString(), + anyString())).thenThrow(failure); + } + + assertThatThrownBy(() -> service.uploadKnowledgeBase(file, null, null)).isSameAs(failure); + + assertThat(service.uploadKnowledgeBase(file, null, null)).containsEntry("duplicate", true); + if (!storageFails) { + verify(storageService).deleteKnowledgeBase("kb/orphan"); + } + verify(producer, never()).sendVectorizeTask(anyLong()); + } + + @Test + @DisplayName("上传调用因中断退出后恢复名额，不影响后续上传") + void shouldReleasePermitWhenInterruptedUploadExits() throws Exception { + service = createService(1); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch exited = new CountDownLatch(1); + AtomicInteger validationCalls = new AtomicInteger(); + doAnswer(invocation -> { + if (validationCalls.incrementAndGet() == 1) { + entered.countDown(); + new CountDownLatch(1).await(10, TimeUnit.SECONDS); + } + return null; + }).when(validationService).validateFile(any(), anyLong(), eq("知识库")); + stubDuplicateUpload(); + ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(1)); + try { + Future upload = executor.submit(() -> { + try { + service.uploadKnowledgeBase(file, null, null); + } finally { + exited.countDown(); + } + }); + assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(upload.cancel(true)).isTrue(); + assertThat(exited.await(5, TimeUnit.SECONDS)).isTrue(); + + assertThat(service.uploadKnowledgeBase(file, null, null)).containsEntry("duplicate", true); + } finally { + executor.shutdownNow(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + } + + @Test + @DisplayName("超限上传沿用 HTTP 200 与 Result 错误码，并返回可重试提示") + void shouldReturnBusyResultFromUploadEndpoint() throws Exception { + service = createService(1); + CountDownLatch entered = new CountDownLatch(1); + CountDownLatch release = new CountDownLatch(1); + AtomicInteger validationCalls = new AtomicInteger(); + doAnswer(invocation -> { + if (validationCalls.incrementAndGet() == 1) { + entered.countDown(); + assertThat(release.await(10, TimeUnit.SECONDS)).isTrue(); + } + return null; + }).when(validationService).validateFile(any(), anyLong(), eq("知识库")); + stubDuplicateUpload(); + MockMvc api = MockMvcBuilders.standaloneSetup(new KnowledgeBaseController(service, + mock(KnowledgeBaseQueryService.class), mock(KnowledgeBaseListService.class), + mock(KnowledgeBaseDeleteService.class))) + .setControllerAdvice(new GlobalExceptionHandler()).build(); + ThreadPoolExecutor executor = new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, + new ArrayBlockingQueue<>(1)); + try { + Future upload = executor.submit(() -> service.uploadKnowledgeBase(file, null, null)); + assertThat(entered.await(5, TimeUnit.SECONDS)).isTrue(); + + api.perform(multipart("/api/knowledgebase/upload").file((MockMultipartFile) file)) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.code").value(8001)) + .andExpect(jsonPath("$.message").value("知识库上传繁忙，请稍后重试")); + + assertThat(validationCalls).hasValue(1); + release.countDown(); + upload.get(5, TimeUnit.SECONDS); + } finally { + release.countDown(); + executor.shutdownNow(); + assertThat(executor.awaitTermination(5, TimeUnit.SECONDS)).isTrue(); + } + } + + private KnowledgeBaseUploadService createService(int maximum) { + KnowledgeBaseUploadProperties properties = new KnowledgeBaseUploadProperties(); + properties.setMaxConcurrent(maximum); + return new KnowledgeBaseUploadService(parseService, persistenceService, storageService, + repository, validationService, hashService, producer, + new KnowledgeBaseUploadLimiter(properties)); + } + + private KnowledgeBaseEntity stubDuplicateUpload() { + KnowledgeBaseEntity existing = new KnowledgeBaseEntity(); + existing.setId(42L); + when(parseService.detectContentType(file)).thenReturn("application/pdf"); + when(hashService.calculateHash(file)).thenReturn("existing-hash"); + when(repository.findByFileHash("existing-hash")).thenReturn(Optional.of(existing)); + when(persistenceService.handleDuplicateKnowledgeBase(existing, "existing-hash")) + .thenReturn(Map.of("duplicate", true)); + return existing; + } +} diff --git a/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadServiceTest.java b/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadServiceTest.java index d03233e3..ad0afc51 100644 --- a/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadServiceTest.java +++ b/app/src/test/java/interview/guide/modules/knowledgebase/service/KnowledgeBaseUploadServiceTest.java @@ -1,8 +1,10 @@ package interview.guide.modules.knowledgebase.service; +import interview.guide.common.config.KnowledgeBaseUploadProperties; import interview.guide.infrastructure.file.FileHashService; import interview.guide.infrastructure.file.FileStorageService; import interview.guide.infrastructure.file.FileValidationService; +import interview.guide.infrastructure.file.KnowledgeBaseUploadLimiter; import interview.guide.modules.knowledgebase.listener.VectorizeStreamProducer; import interview.guide.modules.knowledgebase.model.KnowledgeBaseEntity; import interview.guide.modules.knowledgebase.repository.KnowledgeBaseRepository; @@ -52,7 +54,8 @@ class KnowledgeBaseUploadServiceTest { @BeforeEach void setUp() { service = new KnowledgeBaseUploadService(parseService, persistenceService, storageService, - knowledgeBaseRepository, fileValidationService, fileHashService, vectorizeStreamProducer); + knowledgeBaseRepository, fileValidationService, fileHashService, vectorizeStreamProducer, + new KnowledgeBaseUploadLimiter(new KnowledgeBaseUploadProperties())); } @Test