task-67: 复用 RustFS/MinIO 客户端与 HTTP 连接池
- buildClient 改为 double-check 懒加载缓存共享 MinioClient（volatile 单例）， 同一实例持有同一 OkHttpClient，连接池随实例共享，不再每次操作新建客户端 - supplier 注入路径优先且不写共享缓存，测试隔离不受污染 - 未配置/空白 endpoint 时不创建实例，拒绝操作并保持缓存为空
This commit is contained in:
+218
@@ -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<MeterRegistry> meterRegistryProvider;
|
||||
private ObjectProvider<RustfsDeleteRetryService> 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<Object> 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<Object> 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 路径不写共享缓存");
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user