super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(backend-java): 收口并发资源与恢复链路

Browse Source
This commit is contained in:
黄自达
2026-09-05 14:33:07 +08:00
parent d760838e71
commit 40224d8e56
35 changed files with 917 additions and 154 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java
+3
View File
@@ -38,4 +38,7 @@ public class AppearancePatentProperties {
* Python 回传超时后会直接封口；这个配置处理 Python 慢回传但仍未超时的零头批次。
*/
private int flushPendingMinutes = 1;
/** 单个外观专利源文件最多解析的有效数据行数，防止 POI 用户模型撑爆堆。 */
private int maxParseRows = 50000;
}
backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java
+1 -1
View File
@@ -8,7 +8,7 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
public class BrandCheckProperties {
private String baseUrl = "http://47.110.241.161:16890";
private String path = "/brand_check";
private String token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9";
private String token = "";
private String defaultStrategy = "Terms";
private int connectTimeoutMillis = 10000;
private int readTimeoutMillis = 60000;
backend-java/src/main/java/com/nanri/aiimage/config/HttpClientPool.java
+20 -2
View File
@@ -16,6 +16,20 @@ import java.time.Duration;
public class HttpClientPool {
private static volatile HttpClient sharedHttpClient;
private static volatile long configuredCallTimeoutMillis;
/** 由 Spring 配置属性在启动阶段调用，确保共享客户端使用实际的 connect/call 配置。 */
public static void configure(long connectTimeoutMillis, long callTimeoutMillis) {
configuredCallTimeoutMillis = Math.max(1_000L, callTimeoutMillis);
synchronized (HttpClientPool.class) {
if (sharedHttpClient == null) {
sharedHttpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofMillis(Math.max(1_000L, connectTimeoutMillis)))
.version(HttpClient.Version.HTTP_1_1)
.build();
}
}
}
/** 共享连接池实例：单一 HttpClient 承载全部外部调用的连接复用。 */
public static HttpClient sharedHttpClient() {
@@ -26,7 +40,7 @@ public class HttpClientPool {
synchronized (HttpClientPool.class) {
if (sharedHttpClient == null) {
sharedHttpClient = HttpClient.newBuilder()
.connectTimeout(Duration.ofSeconds(10))
.connectTimeout(Duration.ofMillis(10_000L))
.version(HttpClient.Version.HTTP_1_1)
.build();
}
@@ -36,7 +50,11 @@ public class HttpClientPool {
/** 按 readTimeout（毫秒）创建共享连接池工厂；非法值钳制到最小正数。 */
public static ClientHttpRequestFactory requestFactory(int readTimeoutMillis) {
int safeReadTimeout = Math.max(1, readTimeoutMillis);
long safeReadTimeout = Math.max(1L, readTimeoutMillis);
long callTimeout = configuredCallTimeoutMillis;
if (callTimeout > 0L) {
safeReadTimeout = Math.min(safeReadTimeout, callTimeout);
}
JdkClientHttpRequestFactory factory =
new JdkClientHttpRequestFactory(sharedHttpClient());
factory.setReadTimeout(Duration.ofMillis(safeReadTimeout));
backend-java/src/main/java/com/nanri/aiimage/config/HttpClientProperties.java
+6
View File
@@ -1,5 +1,6 @@
package com.nanri.aiimage.config;
import jakarta.annotation.PostConstruct;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.stereotype.Component;
@@ -31,6 +32,11 @@ public class HttpClientProperties {
private long baseRetryDelayMillis = 500;
/** 钳制后的连接超时：1s-300s。 */
@PostConstruct
void configureSharedHttpClient() {
HttpClientPool.configure(effectiveConnectTimeoutMillis(), effectiveCallTimeoutMillis());
}
public long effectiveConnectTimeoutMillis() {
return clamp(connectTimeoutMillis, 1_000, 300_000);
}
backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java
+48 -17
View File
@@ -59,23 +59,43 @@ public class RequestTraceFilter extends OncePerRequestFilter {
filterChain.doFilter(requestToUse, response);
} finally {
long costMs = System.currentTimeMillis() - start;
log.info(
"request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}",
instanceMetadata.getInstanceId(),
instanceMetadata.getSource(),
instanceMetadata.isStable(),
instanceMetadata.getHostname(),
requestToUse.getMethod(),
requestToUse.getRequestURI(),
response.getStatus(),
remoteAddr,
blankToDash(forwardedHost),
blankToDash(forwardedProto),
blankToDash(forwardedPort),
blankToDash(requestId),
blankToDash(requestToUse.getHeader("User-Agent")),
costMs
);
if (shouldLogAtInfo(requestToUse.getRequestURI(), response.getStatus(), costMs)) {
log.info(
"request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}",
instanceMetadata.getInstanceId(),
instanceMetadata.getSource(),
instanceMetadata.isStable(),
instanceMetadata.getHostname(),
requestToUse.getMethod(),
requestToUse.getRequestURI(),
response.getStatus(),
remoteAddr,
blankToDash(forwardedHost),
blankToDash(forwardedProto),
blankToDash(forwardedPort),
blankToDash(requestId),
blankToDash(requestToUse.getHeader("User-Agent")),
costMs
);
} else {
log.debug(
"request-trace instance={} source={} stable={} host={} method={} uri={} status={} remote={} forwardedHost={} forwardedProto={} forwardedPort={} requestId={} userAgent={} costMs={}",
instanceMetadata.getInstanceId(),
instanceMetadata.getSource(),
instanceMetadata.isStable(),
instanceMetadata.getHostname(),
requestToUse.getMethod(),
requestToUse.getRequestURI(),
response.getStatus(),
remoteAddr,
blankToDash(forwardedHost),
blankToDash(forwardedProto),
blankToDash(forwardedPort),
blankToDash(requestId),
blankToDash(requestToUse.getHeader("User-Agent")),
costMs
);
}
}
}
@@ -100,6 +120,17 @@ public class RequestTraceFilter extends OncePerRequestFilter {
return request;
}
private static boolean shouldLogAtInfo(String uri, int status, long costMs) {
if (status >= 500 || costMs >= 1_000L) {
return true;
}
String normalized = uri == null ? "" : uri.toLowerCase(Locale.ROOT);
return !(normalized.contains("/heartbeat")
|| normalized.contains("/progress")
|| normalized.contains("/poll")
|| normalized.contains("/status"));
}
private static String firstNonBlank(String... values) {
for (String value : values) {
if (value != null && !value.isBlank()) {
backend-java/src/main/java/com/nanri/aiimage/config/TaskFileJobConfig.java
+26 -1
View File
@@ -31,6 +31,7 @@ import com.nanri.aiimage.modules.withdraw.service.WithdrawResultFileJobHandler;
import com.nanri.aiimage.modules.withdraw.service.WithdrawTaskService;
import io.micrometer.core.instrument.MeterRegistry;
import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@@ -40,7 +41,10 @@ import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.Executors;
import java.util.concurrent.Semaphore;
@@ -77,9 +81,30 @@ public class TaskFileJobConfig {
.factory());
}
/**
* transient payload 物理删除使用独立有界线程池，不能与业务虚拟线程执行器共用，
* 避免清理洪峰占满业务任务线程并形成无界在途删除。
*/
@Bean(name = "transientPayloadDeleteExecutor", destroyMethod = "shutdown")
public ExecutorService transientPayloadDeleteExecutor(
@Value("${aiimage.transient-storage.delete-dispatch-pool-size:2}") int poolSize,
@Value("${aiimage.transient-storage.delete-dispatch-queue-capacity:100}") int queueCapacity) {
int workers = Math.max(1, Math.min(poolSize, 16));
int queue = Math.max(1, Math.min(queueCapacity, 10_000));
return new ThreadPoolExecutor(
workers, workers, 0L, TimeUnit.MILLISECONDS,
new ArrayBlockingQueue<>(queue),
runnable -> {
Thread thread = new Thread(runnable, "transient-payload-delete");
thread.setDaemon(true);
return thread;
},
new ThreadPoolExecutor.AbortPolicy());
}
@Bean("taskQueueExecutor")
public TaskExecutor taskQueueExecutor(
ExecutorService taskQueueVirtualThreadExecutor,
@Qualifier("taskQueueVirtualThreadExecutor") ExecutorService taskQueueVirtualThreadExecutor,
@Value("${aiimage.coze-task.max-concurrent:12}") int maxConcurrent,
@Value("${aiimage.coze-task.max-waiting:1000}") int maxWaiting,
ObjectProvider<MeterRegistry> meterRegistryProvider) {
backend-java/src/main/java/com/nanri/aiimage/config/TaskQueueGate.java
+17 -2
View File
@@ -45,14 +45,13 @@ public class TaskQueueGate implements TaskExecutor {
recordRejected("invalid-input");
throw new IllegalArgumentException("task 不能为 null");
}
if (waiting.get() >= maxWaiting) {
if (!tryReserveWaitingSlot()) {
recordRejected("queue-full");
log.warn("[task-queue][gate] waiting queue full, reject submit waiting={} limit={}",
waiting.get(), maxWaiting);
throw new TaskRejectedException("task 等待队列已满，limit=" + maxWaiting
+ ", waiting=" + waiting.get());
}
waiting.incrementAndGet();
long queuedAt = System.nanoTime();
try {
delegate.execute(() -> {
@@ -83,6 +82,22 @@ public class TaskQueueGate implements TaskExecutor {
}
}
/**
* 原子预留一个等待槽位。不能使用“先 get 再 increment”，否则并发提交
* 会同时通过检查，导致等待数量突破 maxWaiting。
*/
private boolean tryReserveWaitingSlot() {
while (true) {
int current = waiting.get();
if (current >= maxWaiting) {
return false;
}
if (waiting.compareAndSet(current, current + 1)) {
return true;
}
}
}
private void recordRejected(String reason) {
MeterRegistry registry = meterRegistry();
if (registry != null) {