diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java index d0cf2fcf..64ee8aa6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java @@ -49,6 +49,11 @@ public class RustfsObjectStorageService { private volatile long failureWindowStartedAtMillis; private volatile long circuitOpenUntilMillis; private volatile OkHttpClient httpClient; + /** + * 懒加载缓存的共享 MinioClient（double-check 单例）：复用同一实例与同一 + * OkHttpClient（连接池随实例共享），避免每次操作创建客户端。 + */ + private volatile MinioClient sharedMinioClient; @Autowired public RustfsObjectStorageService(TransientStorageProperties properties, @@ -275,12 +280,21 @@ public class RustfsObjectStorageService { if (minioClientSupplier != null) { return minioClientSupplier.get(); } - return MinioClient.builder() - .endpoint(properties.getEndpoint()) - .credentials(properties.getAccessKeyId(), properties.getAccessKeySecret()) - .region(properties.getRegion()) - .httpClient(getHttpClient(deadlineNanos)) - .build(); + MinioClient shared = sharedMinioClient; + if (shared != null) { + return shared; + } + synchronized (this) { + if (sharedMinioClient == null) { + sharedMinioClient = MinioClient.builder() + .endpoint(properties.getEndpoint()) + .credentials(properties.getAccessKeyId(), properties.getAccessKeySecret()) + .region(properties.getRegion()) + .httpClient(getHttpClient(deadlineNanos)) + .build(); + } + return sharedMinioClient; + } } OkHttpClient getHttpClient() { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsClientReuseTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsClientReuseTest.java new file mode 100644 index 00000000..b6b7d00a --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/file/service/object/RustfsClientReuseTest.java @@ -0,0 +1,218 @@ +package com.nanri.aiimage.modules.file.service.object; + +import com.nanri.aiimage.config.TransientStorageProperties; +import io.micrometer.core.instrument.MeterRegistry; +import io.micrometer.core.instrument.simple.SimpleMeterRegistry; +import io.minio.MinioClient; +import io.minio.PutObjectArgs; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.ArgumentMatchers; +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.test.util.ReflectionTestUtils; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 67：复用 RustFS/MinIO 客户端与 HTTP 连接池，减少每次操作创建客户端。 + * RustfsObjectStorageService 懒加载缓存共享 MinioClient（double-check 单例）， + * 后续操作复用同一实例与同一 OkHttpClient（连接池随实例共享）， + * 不再每次操作新建客户端；supplier 注入路径优先且不污染共享缓存； + * 未配置/非法配置时不创建实例。 + */ +class RustfsClientReuseTest { + + private TransientStorageProperties properties; + private SimpleMeterRegistry simpleRegistry; + private MinioClient client; + private ObjectProvider meterRegistryProvider; + private ObjectProvider deleteRetryProvider; + private RustfsObjectStorageService service; + + @BeforeEach + @SuppressWarnings("unchecked") + void setUp() { + properties = new TransientStorageProperties(); + simpleRegistry = new SimpleMeterRegistry(); + meterRegistryProvider = mock(ObjectProvider.class); + when(meterRegistryProvider.getIfAvailable()).thenReturn(simpleRegistry); + deleteRetryProvider = mock(ObjectProvider.class); + client = mock(MinioClient.class); + // 默认无 supplier → 走共享 MinioClient 缓存路径 + service = new RustfsObjectStorageService(properties, meterRegistryProvider, deleteRetryProvider, null); + } + + @AfterEach + void tearDown() { + simpleRegistry.clear(); + } + + private static void setConfigured(TransientStorageProperties properties) { + properties.setEndpoint("http://rustfs.local:9000"); + properties.setBucket("bucket"); + properties.setAccessKeyId("ak"); + properties.setAccessKeySecret("sk"); + properties.setRegion("us-east-1"); + } + + private Object sharedMinioClient() { + return ReflectionTestUtils.getField(service, "sharedMinioClient"); + } + + private Object buildClient(long deadlineNanos) { + return ReflectionTestUtils.invokeMethod(service, "buildClient", deadlineNanos); + } + + @Test + void test_task_067_rustfs_normal_default_path() { + // 默认路径：共享缓存懒加载创建 MinioClient，第二次调用复用同一实例。 + setConfigured(properties); + + Object first = buildClient(Long.MAX_VALUE); + Object second = buildClient(Long.MAX_VALUE); + + assertNotNull(first, "首次构建出客户端"); + assertSame(first, second, "复用同一 MinioClient 实例"); + assertSame(first, sharedMinioClient(), "共享缓存字段已填充且一致"); + } + + @Test + void test_task_067_rustfs_normal_multiple_items() throws Exception { + // 批量场景：并发多操作同时构建客户端只产生一个共享实例，无实例爆炸。 + setConfigured(properties); + int concurrency = 16; + CountDownLatch start = new CountDownLatch(1); + Set instances = ConcurrentHashMap.newKeySet(); + Thread[] threads = new Thread[concurrency]; + for (int i = 0; i < concurrency; i++) { + threads[i] = new Thread(() -> { + try { + start.await(2, TimeUnit.SECONDS); + instances.add(buildClient(Long.MAX_VALUE)); + } catch (Exception ignored) { + } + }); + threads[i].setDaemon(true); + threads[i].start(); + } + start.countDown(); + for (Thread thread : threads) { + thread.join(5_000); + } + + assertEquals(1, instances.size(), "并发构建只产生一个共享实例"); + assertNotNull(sharedMinioClient()); + } + + @Test + void test_task_067_rustfs_normal_repeated_operation_is_idempotent() { + // 幂等：重复调用不创建重复客户端，共享实例引用稳定。 + setConfigured(properties); + + Object first = buildClient(Long.MAX_VALUE); + Object second = buildClient(Long.MAX_VALUE); + Object third = buildClient(Long.MAX_VALUE); + + assertSame(first, second); + assertSame(second, third); + assertSame(first, sharedMinioClient()); + } + + @Test + void test_task_067_rustfs_boundary_empty_input() { + // 空输入：未配置时操作直接拒绝，不创建任何客户端实例。 + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> service.uploadText("k", "v")); + org.junit.jupiter.api.Assertions.assertTrue(ex.getMessage().contains("not configured")); + assertNull(sharedMinioClient(), "未配置不创建客户端"); + } + + @Test + void test_task_067_rustfs_boundary_single_item() { + // 单元素：单次构建创建一次并缓存，不依赖批量路径。 + setConfigured(properties); + + Object built = buildClient(Long.MAX_VALUE); + + assertNotNull(built); + assertSame(built, sharedMinioClient()); + } + + @Test + void test_task_067_rustfs_boundary_limit_and_overflow() throws Exception { + // 上限/超限：更大并发下仍只创建一个共享实例； + // HTTP 客户端（含连接池）同样复用同一实例。 + setConfigured(properties); + int concurrency = 32; + CountDownLatch start = new CountDownLatch(1); + Set instances = ConcurrentHashMap.newKeySet(); + Thread[] threads = new Thread[concurrency]; + for (int i = 0; i < concurrency; i++) { + threads[i] = new Thread(() -> { + try { + start.await(2, TimeUnit.SECONDS); + instances.add(buildClient(Long.MAX_VALUE)); + } catch (Exception ignored) { + } + }); + threads[i].setDaemon(true); + threads[i].start(); + } + start.countDown(); + for (Thread thread : threads) { + thread.join(5_000); + } + + assertEquals(1, instances.size(), "高并发下不产生客户端实例爆炸"); + assertSame(service.getHttpClient(), service.getHttpClient(), "HTTP 客户端复用同一实例（连接池共享）"); + } + + @Test + void test_task_067_rustfs_invalid_input_rejected() { + // 非法参数：endpoint 为空白时视为未配置，拒绝操作且不创建客户端。 + properties.setEndpoint(" "); + properties.setBucket("bucket"); + properties.setAccessKeyId("ak"); + properties.setAccessKeySecret("sk"); + + IllegalStateException ex = assertThrows(IllegalStateException.class, + () -> service.uploadText("k", "v")); + org.junit.jupiter.api.Assertions.assertTrue(ex.getMessage().contains("not configured")); + assertNull(sharedMinioClient()); + } + + @Test + void test_task_067_rustfs_dependency_failure_releases_resources() throws Exception { + // 依赖失败：supplier 注入路径优先且失败后可恢复， + // 不污染共享缓存（sharedMinioClient 保持未初始化）。 + RustfsObjectStorageService supplierService = new RustfsObjectStorageService( + properties, meterRegistryProvider, deleteRetryProvider, () -> client); + setConfigured(properties); + properties.setUploadMaxRetries(1); + when(client.putObject(ArgumentMatchers.any(PutObjectArgs.class))) + .thenThrow(new java.io.IOException("rustfs down")); + assertThrows(Exception.class, () -> supplierService.uploadText("k", "v")); + + when(client.putObject(ArgumentMatchers.any(PutObjectArgs.class))) + .thenReturn(mock(io.minio.ObjectWriteResponse.class)); + when(client.statObject(ArgumentMatchers.any(io.minio.StatObjectArgs.class))).thenReturn(null); + assertEquals("k", supplierService.uploadText("k", "v")); + verify(client, times(2)).putObject(ArgumentMatchers.any(PutObjectArgs.class)); + assertNull(ReflectionTestUtils.getField(supplierService, "sharedMinioClient"), + "supplier 路径不写共享缓存"); + } +}