From 15adcfff428a61bbd56d97cff0b219aeb356f688 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=99=88=E5=AD=90=E9=BB=98?= <925456043@qq.com> Date: Mon, 3 Aug 2026 11:32:25 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=20MinerU=20=E5=BC=82?= =?UTF-8?q?=E6=AD=A5=E4=BB=BB=E5=8A=A1=E6=8F=90=E4=BA=A4=E6=8E=A2=E6=B5=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 使用健康接口预检服务可用性并缓存短期故障 - 限制任务提交总时长并快速处理服务端错误 - 补齐 Starter 配置传递与核心回归测试 --- .../document/core/mineru/MineruClient.java | 134 +++++++++++- .../core/mineru/MineruProperties.java | 19 ++ .../MineruDocumentParseServiceTest.java | 201 ++++++++++++++++++ .../CommonMineruDocumentProperties.java | 19 ++ .../mineru/MineruPdfAutoConfiguration.java | 1 + .../pptx/MineruPptxAutoConfiguration.java | 1 + .../xlsx/MineruXlsxAutoConfiguration.java | 1 + 7 files changed, 373 insertions(+), 3 deletions(-) diff --git a/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruClient.java b/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruClient.java index be2fb6a..216921c 100644 --- a/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruClient.java +++ b/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruClient.java @@ -5,6 +5,7 @@ import com.easyagents.core.util.StringUtil; import com.easyagents.document.core.exception.DocumentParseException; import com.easyagents.document.core.entity.ParseFile; import com.easyagents.document.core.entity.ParseRequest; +import okhttp3.Call; import okhttp3.MediaType; import okhttp3.MultipartBody; import okhttp3.OkHttpClient; @@ -29,10 +30,17 @@ import java.util.concurrent.TimeUnit; public class MineruClient { private static final MediaType DEFAULT_MEDIA_TYPE = MediaType.parse("application/octet-stream"); + private static final int DEFAULT_SUBMIT_TIMEOUT_MS = 120000; + private static final int AVAILABILITY_PROBE_TIMEOUT_MS = 3000; + private static final long AVAILABILITY_CACHE_TTL_MS = 5000L; private final String baseUrl; private final OkHttpClient okHttpClient; private final MineruMapper mineruMapper; + private final int submitTimeoutMs; + private final Object availabilityProbeMonitor = new Object(); + private volatile long availabilityCacheDeadlineMs; + private volatile String availabilityFailureMessage; /** * 创建客户端。 @@ -66,6 +74,7 @@ public class MineruClient { this.baseUrl = normalizeBaseUrl(properties.getBaseUrl()); this.okHttpClient = okHttpClient; this.mineruMapper = mineruMapper; + this.submitTimeoutMs = positiveOrDefault(properties.getSubmitTimeoutMs(), DEFAULT_SUBMIT_TIMEOUT_MS); } /** @@ -85,7 +94,10 @@ public class MineruClient { * @return 原始任务状态 */ public MineruTaskStatus submit(ParseRequest request) { - return mineruMapper.toTaskStatus(executeJsonMultipart("/tasks", request, buildAsyncFormFields(request))); + assertServiceAvailable(); + return mineruMapper.toTaskStatus( + executeJsonMultipart("/tasks", request, buildAsyncFormFields(request), submitTimeoutMs) + ); } /** @@ -131,6 +143,22 @@ public class MineruClient { } protected JSONObject executeJsonMultipart(String path, ParseRequest request, Map> fields) { + return executeJsonMultipart(path, request, fields, 0); + } + + /** + * 执行带整次调用超时的 Multipart JSON 请求。 + * + * @param path 接口路径 + * @param request 解析请求 + * @param fields 表单字段 + * @param callTimeoutMs 整次调用超时时间，单位毫秒；小于等于 0 时沿用客户端阶段超时 + * @return JSON 响应 + */ + protected JSONObject executeJsonMultipart(String path, + ParseRequest request, + Map> fields, + long callTimeoutMs) { MultipartBody.Builder formBuilder = new MultipartBody.Builder().setType(MultipartBody.FORM); appendFiles(formBuilder, request.getFiles()); appendStringFields(formBuilder, fields); @@ -138,7 +166,9 @@ public class MineruClient { .url(baseUrl + path) .post(formBuilder.build()) .build(); - return executeJsonRequest(path, httpRequest); + return callTimeoutMs > 0 + ? executeJsonRequest(path, httpRequest, callTimeoutMs) + : executeJsonRequest(path, httpRequest); } protected JSONObject executeJsonGet(String path) { @@ -147,7 +177,28 @@ public class MineruClient { } protected JSONObject executeJsonRequest(String path, Request request) { - try (Response response = okHttpClient.newCall(request).execute()) { + return executeJsonRequest(path, request, 0); + } + + /** + * 执行 JSON 请求，并可限制连接、写入和读取在内的整次调用时长。 + * + * @param path 接口路径 + * @param request HTTP 请求 + * @param callTimeoutMs 整次调用超时时间，单位毫秒；小于等于 0 时不额外限制 + * @return JSON 响应 + */ + protected JSONObject executeJsonRequest(String path, Request request, long callTimeoutMs) { + Call call = okHttpClient.newCall(request); + if (callTimeoutMs > 0) { + // OkHttp 的 Call timeout 到期后会取消底层请求，避免线程长期阻塞在文件上传或响应等待。 + call.timeout().timeout(callTimeoutMs, TimeUnit.MILLISECONDS); + } + try (Response response = call.execute()) { + if (response.code() >= 500) { + // 服务端错误在响应头到达时立即失败，避免异常响应体未结束时继续占用提交线程。 + throw buildHttpException(path, response.code(), new byte[0]); + } ResponseBody body = response.body(); String bodyText = body == null ? "" : body.string(); if (!response.isSuccessful()) { @@ -163,6 +214,83 @@ public class MineruClient { } } + /** + * 在上传文件前探测 MinerU 网关是否可用，避免服务已返回 5xx 时仍传输大文件。 + * + *

探测结果短暂缓存，限制批量导入期间的额外请求量。健康检查返回 4xx 说明网关仍可达， + * 实际任务接口会继续完成业务校验。

+ * + * @throws DocumentParseException 网关返回 5xx 或探测请求失败 + */ + private void assertServiceAvailable() { + long now = System.currentTimeMillis(); + if (now < availabilityCacheDeadlineMs) { + throwCachedAvailabilityFailure(); + return; + } + synchronized (availabilityProbeMonitor) { + now = System.currentTimeMillis(); + if (now < availabilityCacheDeadlineMs) { + throwCachedAvailabilityFailure(); + return; + } + String healthPath = "/health"; + Request request = new Request.Builder().url(baseUrl + healthPath).get().build(); + Call call = okHttpClient.newCall(request); + call.timeout().timeout( + Math.min(submitTimeoutMs, AVAILABILITY_PROBE_TIMEOUT_MS), + TimeUnit.MILLISECONDS + ); + try (Response response = call.execute()) { + if (response.code() >= 500) { + cacheAvailabilityFailure( + "MinerU service unavailable: path=" + healthPath + ", status=" + response.code(), + now + ); + throw new DocumentParseException(availabilityFailureMessage); + } + availabilityFailureMessage = null; + availabilityCacheDeadlineMs = now + AVAILABILITY_CACHE_TTL_MS; + } catch (IOException exception) { + cacheAvailabilityFailure("MinerU service unavailable: availability probe failed", now); + throw new DocumentParseException(availabilityFailureMessage, exception); + } + } + } + + /** + * 缓存 MinerU 不可用状态。 + * + * @param message 失败信息 + * @param detectedAtMs 检测时间，单位毫秒 + */ + private void cacheAvailabilityFailure(String message, long detectedAtMs) { + availabilityFailureMessage = message; + availabilityCacheDeadlineMs = detectedAtMs + AVAILABILITY_CACHE_TTL_MS; + } + + /** + * 若缓存记录为不可用则抛出稳定异常。 + * + * @throws DocumentParseException MinerU 仍处于不可用缓存窗口 + */ + private void throwCachedAvailabilityFailure() { + if (availabilityFailureMessage != null) { + throw new DocumentParseException(availabilityFailureMessage); + } + } + + /** + * 返回正整数配置，非法配置回退到缺省值。 + * + * @param value 配置值 + * @param defaultValue 缺省值 + * @return 可用的正整数 + */ + private int positiveOrDefault(Integer value, int defaultValue) { + return value == null || value <= 0 ? defaultValue : value; + } + private void appendFiles(MultipartBody.Builder formBuilder, List files) { if (files == null || files.isEmpty()) { throw new IllegalArgumentException("Parse request must contain at least one file"); diff --git a/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruProperties.java b/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruProperties.java index d45ab20..1c70fa8 100644 --- a/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruProperties.java +++ b/easy-agents-document/easy-agents-document-core/src/main/java/com/easyagents/document/core/mineru/MineruProperties.java @@ -16,6 +16,7 @@ public class MineruProperties { private Integer connectTimeoutMs = 3000; private Integer readTimeoutMs = 600000; private Integer writeTimeoutMs = 600000; + private Integer submitTimeoutMs = 120000; private Integer pollIntervalMs = 1000; private Integer resultTimeoutMs = 1800000; private String defaultBackend = "vlm-http-client"; @@ -56,6 +57,24 @@ public class MineruProperties { this.writeTimeoutMs = writeTimeoutMs; } + /** + * 获取异步任务提交总超时时间。 + * + * @return 提交总超时时间，单位毫秒 + */ + public Integer getSubmitTimeoutMs() { + return submitTimeoutMs; + } + + /** + * 设置异步任务提交总超时时间。 + * + * @param submitTimeoutMs 提交总超时时间，单位毫秒 + */ + public void setSubmitTimeoutMs(Integer submitTimeoutMs) { + this.submitTimeoutMs = submitTimeoutMs; + } + public Integer getPollIntervalMs() { return pollIntervalMs; } diff --git a/easy-agents-document/easy-agents-document-core/src/test/java/com/easyagents/document/core/mineru/MineruDocumentParseServiceTest.java b/easy-agents-document/easy-agents-document-core/src/test/java/com/easyagents/document/core/mineru/MineruDocumentParseServiceTest.java index cdcaeb8..916e47a 100644 --- a/easy-agents-document/easy-agents-document-core/src/test/java/com/easyagents/document/core/mineru/MineruDocumentParseServiceTest.java +++ b/easy-agents-document/easy-agents-document-core/src/test/java/com/easyagents/document/core/mineru/MineruDocumentParseServiceTest.java @@ -6,14 +6,30 @@ import com.easyagents.document.core.entity.ParseRequest; import com.easyagents.document.core.entity.ParseResponse; import com.easyagents.document.core.entity.ParseTaskInfo; import com.easyagents.document.core.entity.ParseTaskStatus; +import com.easyagents.document.core.exception.DocumentParseException; +import okhttp3.MediaType; +import okhttp3.OkHttpClient; +import okhttp3.Protocol; import okhttp3.Request; +import okhttp3.Response; +import okhttp3.ResponseBody; import okio.Buffer; +import okio.BufferedSource; +import okio.Okio; +import okio.Source; +import okio.Timeout; import org.junit.Assert; import org.junit.Test; import java.io.ByteArrayOutputStream; +import java.io.BufferedReader; import java.io.IOException; +import java.io.InputStreamReader; +import java.net.ServerSocket; +import java.net.Socket; import java.nio.charset.StandardCharsets; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import java.util.zip.ZipEntry; import java.util.zip.ZipOutputStream; @@ -89,6 +105,191 @@ public class MineruDocumentParseServiceTest { Assert.assertTrue(client.lastMultipartBody.contains("\r\nen\r\n")); } + /** + * 验证异步任务提交总超时会取消无响应的 HTTP 调用。 + * + * @throws Exception 本地测试套接字异常 + */ + @Test + public void submitShouldCancelUnresponsiveCallAtConfiguredTimeout() throws Exception { + AtomicReference acceptedSocket = new AtomicReference(); + try (ServerSocket serverSocket = new ServerSocket(0)) { + Thread serverThread = new Thread(() -> { + try { + Socket socket = serverSocket.accept(); + acceptedSocket.set(socket); + while (socket.getInputStream().read() >= 0) { + // 持续读取请求但不返回响应，模拟 MinerU 提交接口失去响应。 + } + } catch (IOException ignore) { + // 客户端超时取消或测试关闭套接字后结束服务线程。 + } + }, "mineru-submit-timeout-test-server"); + serverThread.setDaemon(true); + serverThread.start(); + + MineruProperties properties = defaultProperties(); + properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort()); + properties.setSubmitTimeoutMs(200); + properties.setConnectTimeoutMs(5000); + properties.setReadTimeoutMs(5000); + properties.setWriteTimeoutMs(5000); + MineruClient client = new MineruClient(properties, new MineruMapper(properties)); + + long startedAt = System.currentTimeMillis(); + try { + client.submit(buildRequest()); + Assert.fail("Expected MinerU submit timeout"); + } catch (DocumentParseException expected) { + long elapsed = System.currentTimeMillis() - startedAt; + Assert.assertTrue("Submit call should be cancelled promptly, elapsed=" + elapsed, elapsed < 2000); + } finally { + Socket socket = acceptedSocket.get(); + if (socket != null) { + socket.close(); + } + serverThread.join(1000); + } + } + } + + /** + * 验证 MinerU 网关返回 5xx 时会在上传文件前立即失败，并复用短期不可用缓存。 + * + * @throws Exception 本地测试套接字异常 + */ + @Test + public void submitShouldFailBeforeMultipartUploadWhenProbeReturnsServerError() throws Exception { + AtomicInteger requestCount = new AtomicInteger(); + AtomicReference requestLine = new AtomicReference(); + Thread serverThread; + try (ServerSocket serverSocket = new ServerSocket(0)) { + serverThread = new Thread(() -> { + while (!serverSocket.isClosed()) { + try (Socket socket = serverSocket.accept(); + BufferedReader reader = new BufferedReader( + new InputStreamReader(socket.getInputStream(), StandardCharsets.UTF_8))) { + requestCount.incrementAndGet(); + requestLine.set(reader.readLine()); + String header; + while ((header = reader.readLine()) != null && !header.isEmpty()) { + // 读取完整请求头后再返回网关错误。 + } + byte[] body = "Service Unavailable".getBytes(StandardCharsets.UTF_8); + String responseHeaders = "HTTP/1.1 503 Service Unavailable\r\n" + + "Content-Type: text/plain\r\n" + + "Content-Length: " + body.length + "\r\n" + + "Connection: close\r\n\r\n"; + socket.getOutputStream().write(responseHeaders.getBytes(StandardCharsets.UTF_8)); + socket.getOutputStream().write(body); + socket.getOutputStream().flush(); + } catch (IOException ignore) { + // 测试结束关闭 ServerSocket 后退出服务线程。 + } + } + }, "mineru-availability-probe-test-server"); + serverThread.setDaemon(true); + serverThread.start(); + + MineruProperties properties = defaultProperties(); + properties.setBaseUrl("http://127.0.0.1:" + serverSocket.getLocalPort()); + MineruClient client = new MineruClient(properties, new MineruMapper(properties)); + + long startedAt = System.currentTimeMillis(); + for (int attempt = 0; attempt < 2; attempt++) { + try { + client.submit(buildRequest()); + Assert.fail("Expected MinerU availability failure"); + } catch (DocumentParseException expected) { + Assert.assertTrue(expected.getMessage().contains("status=503")); + } + } + long elapsed = System.currentTimeMillis() - startedAt; + Assert.assertTrue("Server error should fail promptly, elapsed=" + elapsed, elapsed < 2000); + } + serverThread.join(1000); + + Assert.assertEquals("Cached failure should avoid repeated probes", 1, requestCount.get()); + Assert.assertTrue(requestLine.get().startsWith("GET /health HTTP/1.1")); + } + + /** + * 验证任务接口返回 5xx 后不会等待或读取异常响应体。 + */ + @Test + public void submitShouldThrowServerErrorWithoutReadingResponseBody() { + AtomicInteger requestCount = new AtomicInteger(); + AtomicReference errorBodyRead = new AtomicReference(false); + OkHttpClient httpClient = new OkHttpClient.Builder() + .addInterceptor(chain -> { + int currentRequest = requestCount.incrementAndGet(); + Response.Builder responseBuilder = new Response.Builder() + .request(chain.request()) + .protocol(Protocol.HTTP_1_1); + if (currentRequest == 1) { + return responseBuilder + .code(200) + .message("OK") + .body(ResponseBody.create((MediaType) null, new byte[0])) + .build(); + } + ResponseBody trackingBody = new ResponseBody() { + + private final BufferedSource source = Okio.buffer(new Source() { + + @Override + public long read(Buffer sink, long byteCount) { + errorBodyRead.set(true); + return -1; + } + + @Override + public Timeout timeout() { + return Timeout.NONE; + } + + @Override + public void close() { + // 无底层资源需要关闭。 + } + }); + + @Override + public MediaType contentType() { + return MediaType.parse("text/plain"); + } + + @Override + public long contentLength() { + return 19; + } + + @Override + public BufferedSource source() { + return source; + } + }; + return responseBuilder + .code(500) + .message("Internal Server Error") + .body(trackingBody) + .build(); + }) + .build(); + MineruProperties properties = defaultProperties(); + MineruClient client = new MineruClient(properties, httpClient, new MineruMapper(properties)); + + try { + client.submit(buildRequest()); + Assert.fail("Expected MinerU server error"); + } catch (DocumentParseException expected) { + Assert.assertTrue(expected.getMessage().contains("status=500")); + } + + Assert.assertEquals(2, requestCount.get()); + Assert.assertFalse("5xx response body should not be read", errorBodyRead.get()); + } + private ParseRequest buildRequest() { ParseRequest request = new ParseRequest(); request.addFile(ParseFile.of("demo.pptx", "ppt".getBytes(StandardCharsets.UTF_8))); diff --git a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/mineru/CommonMineruDocumentProperties.java b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/mineru/CommonMineruDocumentProperties.java index 1e26e56..41f9537 100644 --- a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/mineru/CommonMineruDocumentProperties.java +++ b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/mineru/CommonMineruDocumentProperties.java @@ -19,6 +19,7 @@ public class CommonMineruDocumentProperties { private Integer connectTimeoutMs = 3000; private Integer readTimeoutMs = 600000; private Integer writeTimeoutMs = 600000; + private Integer submitTimeoutMs = 120000; private Integer pollIntervalMs = 1000; private Integer resultTimeoutMs = 1800000; private String defaultBackend = "vlm-http-client"; @@ -59,6 +60,24 @@ public class CommonMineruDocumentProperties { this.writeTimeoutMs = writeTimeoutMs; } + /** + * 获取异步任务提交总超时时间。 + * + * @return 提交总超时时间，单位毫秒 + */ + public Integer getSubmitTimeoutMs() { + return submitTimeoutMs; + } + + /** + * 设置异步任务提交总超时时间。 + * + * @param submitTimeoutMs 提交总超时时间，单位毫秒 + */ + public void setSubmitTimeoutMs(Integer submitTimeoutMs) { + this.submitTimeoutMs = submitTimeoutMs; + } + public Integer getPollIntervalMs() { return pollIntervalMs; } diff --git a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pdf/mineru/MineruPdfAutoConfiguration.java b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pdf/mineru/MineruPdfAutoConfiguration.java index 9cdcc74..be723e6 100644 --- a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pdf/mineru/MineruPdfAutoConfiguration.java +++ b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pdf/mineru/MineruPdfAutoConfiguration.java @@ -105,6 +105,7 @@ public class MineruPdfAutoConfiguration { mineruProperties.setConnectTimeoutMs(useCommon ? commonProperties.getConnectTimeoutMs() : null); mineruProperties.setReadTimeoutMs(useCommon ? commonProperties.getReadTimeoutMs() : null); mineruProperties.setWriteTimeoutMs(useCommon ? commonProperties.getWriteTimeoutMs() : null); + mineruProperties.setSubmitTimeoutMs(useCommon ? commonProperties.getSubmitTimeoutMs() : null); mineruProperties.setPollIntervalMs(useCommon ? commonProperties.getPollIntervalMs() : null); mineruProperties.setResultTimeoutMs(useCommon ? commonProperties.getResultTimeoutMs() : null); mineruProperties.setDefaultBackend(useCommon ? commonProperties.getDefaultBackend() : null); diff --git a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pptx/MineruPptxAutoConfiguration.java b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pptx/MineruPptxAutoConfiguration.java index c367776..a2fbab1 100644 --- a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pptx/MineruPptxAutoConfiguration.java +++ b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/pptx/MineruPptxAutoConfiguration.java @@ -53,6 +53,7 @@ public class MineruPptxAutoConfiguration { mineruProperties.setConnectTimeoutMs(properties.getConnectTimeoutMs()); mineruProperties.setReadTimeoutMs(properties.getReadTimeoutMs()); mineruProperties.setWriteTimeoutMs(properties.getWriteTimeoutMs()); + mineruProperties.setSubmitTimeoutMs(properties.getSubmitTimeoutMs()); mineruProperties.setPollIntervalMs(properties.getPollIntervalMs()); mineruProperties.setResultTimeoutMs(properties.getResultTimeoutMs()); mineruProperties.setDefaultBackend(properties.getDefaultBackend()); diff --git a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/xlsx/MineruXlsxAutoConfiguration.java b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/xlsx/MineruXlsxAutoConfiguration.java index 848e7a2..e580fa4 100644 --- a/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/xlsx/MineruXlsxAutoConfiguration.java +++ b/easy-agents-spring-boot-starter/src/main/java/com/easyagents/spring/boot/document/xlsx/MineruXlsxAutoConfiguration.java @@ -53,6 +53,7 @@ public class MineruXlsxAutoConfiguration { mineruProperties.setConnectTimeoutMs(properties.getConnectTimeoutMs()); mineruProperties.setReadTimeoutMs(properties.getReadTimeoutMs()); mineruProperties.setWriteTimeoutMs(properties.getWriteTimeoutMs()); + mineruProperties.setSubmitTimeoutMs(properties.getSubmitTimeoutMs()); mineruProperties.setPollIntervalMs(properties.getPollIntervalMs()); mineruProperties.setResultTimeoutMs(properties.getResultTimeoutMs()); mineruProperties.setDefaultBackend(properties.getDefaultBackend());