修复商品风险多店铺问题

2026-04-30 21:31:11 +08:00
parent b073bbe97d
commit 01bae23df5
17 changed files with 1392 additions and 46 deletions
@@ -5,18 +5,24 @@ import io.minio.GetObjectArgs;
 import io.minio.MinioClient;
 import io.minio.PutObjectArgs;
 import io.minio.RemoveObjectArgs;
+import io.minio.StatObjectArgs;
 import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import okhttp3.OkHttpClient;
 import org.springframework.stereotype.Service;

 import java.io.ByteArrayInputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Objects;
+import java.util.concurrent.TimeUnit;

@Service
@RequiredArgsConstructor
+@Slf4j
 public class RustfsObjectStorageService {

    private final TransientStorageProperties properties;
+    private volatile OkHttpClient httpClient;

    public boolean isConfigured() {
        return notBlank(properties.getEndpoint())
@@ -29,18 +35,28 @@ public class RustfsObjectStorageService {
        if (!isConfigured()) {
            throw new IllegalStateException("transient storage is not configured");
        }
-        try (ByteArrayInputStream stream = new ByteArrayInputStream(
-                Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8))) {
-            buildClient().putObject(PutObjectArgs.builder()
-                    .bucket(properties.getBucket())
-                    .object(objectKey)
-                    .stream(stream, stream.available(), -1)
-                    .contentType("application/json")
-                    .build());
-            return objectKey;
-        } catch (Exception ex) {
-            throw new IllegalStateException("failed to upload payload to transient storage", ex);
+        byte[] bytes = Objects.requireNonNullElse(content, "").getBytes(StandardCharsets.UTF_8);
+        Exception last = null;
+        int maxRetries = Math.max(1, properties.getUploadMaxRetries());
+        for (int attempt = 1; attempt <= maxRetries; attempt++) {
+            try (ByteArrayInputStream stream = new ByteArrayInputStream(bytes)) {
+                buildClient().putObject(PutObjectArgs.builder()
+                        .bucket(properties.getBucket())
+                        .object(objectKey)
+                        .stream(stream, bytes.length, -1)
+                        .contentType("application/json")
+                        .build());
+                verifyObjectVisible(objectKey);
+                return objectKey;
+            } catch (Exception ex) {
+                last = ex;
+                if (attempt < maxRetries) {
+                    log.warn("[rustfs] upload failed, retrying objectKey={} attempt={}/{}", objectKey, attempt, maxRetries, ex);
+                    sleepQuietly(500L * attempt);
+                }
+            }
        }
+        throw new IllegalStateException("failed to upload payload to transient storage", last);
    }

    public String readObjectAsString(String objectKey) {
@@ -76,9 +92,57 @@ public class RustfsObjectStorageService {
                .endpoint(properties.getEndpoint())
                .credentials(properties.getAccessKeyId(), properties.getAccessKeySecret())
                .region(properties.getRegion())
+                .httpClient(getHttpClient())
                .build();
    }

+    private OkHttpClient getHttpClient() {
+        OkHttpClient current = httpClient;
+        if (current != null) {
+            return current;
+        }
+        synchronized (this) {
+            if (httpClient == null) {
+                httpClient = new OkHttpClient.Builder()
+                        .connectTimeout(Math.max(1, properties.getConnectTimeoutSeconds()), TimeUnit.SECONDS)
+                        .readTimeout(Math.max(1, properties.getReadTimeoutSeconds()), TimeUnit.SECONDS)
+                        .writeTimeout(Math.max(1, properties.getWriteTimeoutSeconds()), TimeUnit.SECONDS)
+                        .build();
+            }
+            return httpClient;
+        }
+    }
+
+    private void verifyObjectVisible(String objectKey) {
+        RuntimeException last = null;
+        for (int attempt = 1; attempt <= 3; attempt++) {
+            try {
+                buildClient().statObject(StatObjectArgs.builder()
+                        .bucket(properties.getBucket())
+                        .object(objectKey)
+                        .build());
+                return;
+            } catch (Exception ex) {
+                last = new IllegalStateException("transient payload is not visible after upload: " + objectKey, ex);
+                if (attempt < 3) {
+                    sleepQuietly(100L * attempt);
+                }
+            }
+        }
+        throw last == null
+                ? new IllegalStateException("transient payload is not visible after upload: " + objectKey)
+                : last;
+    }
+
+    private void sleepQuietly(long millis) {
+        try {
+            Thread.sleep(millis);
+        } catch (InterruptedException ex) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("interrupted while verifying transient payload upload", ex);
+        }
+    }
+
    private boolean notBlank(String value) {
        return value != null && !value.isBlank();
    }