用 RestTemplate 流式调用文本接口时按有没有输出决定是否重试

《炊想》的后端是 Spring Boot 2.7 + JDK 8，核心一步是调一个 OpenAI 兼容的文本接口，让它按提示词写出一整份菜谱 JSON（食材、步骤、搭配、营养，约 1600 个 token）。上游每秒吐十几个 token，一份写完要 60～90 秒。

10 月 12 日中午，这个上游连续半个多小时很不稳定：有时一开口就回「当前模型访问量较大，请稍后重试」，有时干脆 30 秒一个字都不写。翻日志发现，后端好几次在明明还有一大半时间的情况下就判了失败，用户在小程序里看到「搭配师走神了」。

这篇记录当天对重试策略的修改：为什么要流式读取、读超时到底在量什么、怎么在 RestTemplate 里知道「上游到底写没写」、退避怎么和总时限配合，以及这些分支怎么用 MockRestServiceServer 测出来。最后附带一个同一天修的、看起来很蠢的解析问题。

一、先说背景：为什么要流式读

最早的版本是普通请求：发出去，等上游把整段 JSON 写完再一次性读回来。这时 SimpleClientHttpRequestFactory 的 readTimeout 量的其实是「整份必须在多少秒内写完」——上游在写的过程中不会发任何字节，连接上一直是静默的。

10 月 5 日就吃过这个亏：读超时设的 75 秒，两次生成都在 75 秒处断开，重试一次又是 75 秒，小程序在 nginx 的 120 秒处拿到 504。

改成流式（请求体里带 "stream": true，上游按 SSE 一段一段回 data: {...}）以后，读超时的含义变了：HttpURLConnection 的读超时是每次 read 最多阻塞多久，只要上游隔一会儿写一段，就不会超时。于是两个时间可以分开管：

读超时交给 RestTemplate，总时限要自己掐：每读一行检查一次有没有过 deadline。

static RestTemplate build(int connectTimeoutMs, int readTimeoutMs) {
    SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory();
    factory.setConnectTimeout(Math.max(1000, connectTimeoutMs));
    factory.setReadTimeout(Math.max(1000, readTimeoutMs));   // 文本接口传的是 30 秒的空闲超时
    return new RestTemplate(factory);
}

流式读取用 RestTemplate.execute 自己写请求体、自己读响应流，不让它把整个响应读成字符串：

content = restTemplate.execute(url, HttpMethod.POST, request -> {
    request.getHeaders().putAll(headers);
    try (OutputStream out = request.getBody()) {
        out.write(body);
    }
}, response -> readContent(response, deadline, gotOutput));

readContent 按 Content-Type 分流：text/event-stream 就逐行读 data:，把 choices[0].delta.content 拼起来，遇到 data: [DONE] 结束；上游没按流式回（有的兼容接口会忽略 stream）就整段解析 choices[0].message.content。

二、失败有好几种，处理方式不一样

接口调用失败统一抛一个 AiCallException，带 retryable 和 HTTP 状态码。原来的重试规则只有一条：

只在开头 20 秒内就失败的，才重试一次。

道理是：写到一半甚至写完才失败（比如 JSON 解析不了），已经花掉六七十秒，再来一遍必然超过总时限，重试只是让用户多等。

10 月 10 日上线后先碰到一种例外：上游在第 22 秒才回「访问量较大」——它根本没开始写，只是排了 22 秒队才拒绝。按老规则过了 20 秒窗口，直接失败。当时加了一条「上游说忙」的分支：只要总时限还剩 95 秒（一份菜谱一般 60～90 秒写完），就隔 2 秒再试，最多 3 次。

10 月 12 日又碰到两种新情况，把规则彻底理了一遍：

第二行是这次的重点。日志长这样（时间换成了相对秒数）：

第 34 秒  文本上游第 1 次失败（已用 34206 ms，来不及重试）: 文本上游连接/读取超时:
          I/O error on POST request for ".../chat/completions": Read timed out

总时限 140 秒，才用了 34 秒，还剩 106 秒，足够再写一份；但因为过了 20 秒窗口，被当成「写到一半断了」直接判死。问题在于：光看一个 Read timed out，分不清是没开始写还是写到一半。

三、怎么知道上游到底写没写

RestTemplate 把响应读取里抛出的 IOException（包括 SocketTimeoutException）包成 ResourceAccessException 抛出来，到 catch 的时候，读到了多少已经无从知道。只能在读的过程中自己记下来。

第一反应是在 client 上加个字段 volatile boolean gotOutput——不行，这个 client 是单例，好几个生成任务在不同线程里同时调，字段会互相覆盖。进度必须是每次调用自己的局部状态，所以在 callRaw 里开一个只有一个元素的数组，传进读取方法：

final boolean[] gotOutput = {false};
String content;
try {
    content = restTemplate.execute(url, HttpMethod.POST, request -> { /* 写请求体 */ },
            response -> readContent(response, deadline, gotOutput));
} catch (HttpStatusCodeException ex) {
    int status = ex.getRawStatusCode();
    throw new AiCallException("文本上游返回 HTTP " + status + ": " + Texts.truncate(ex.getResponseBodyAsString(), 200),
            status >= 500, status, ex);
} catch (ResourceAccessException ex) {
    AiCallException failed = new AiCallException("文本上游连接/读取超时" + (gotOutput[0] ? "" : "（还没开始输出）")
            + ": " + ex.getMessage(), true, -1, ex);
    throw gotOutput[0] ? failed : failed.markBeforeOutput();
}

lambda 里不能给外层局部变量重新赋值（必须是 effectively final），所以用数组当个可变的盒子；也可以用 AtomicBoolean，这里只在同一个线程里读写，数组就够了。

读 SSE 的循环里，收到第一段正文或推理内容时置位：

while ((line = reader.readLine()) != null) {
    checkDeadline(deadline);
    if (!line.startsWith("data:")) continue;
    String data = line.substring(5).trim();
    if (data.isEmpty()) continue;
    if ("[DONE]".equals(data)) break;
    JsonNode chunk = objectMapper.readTree(data);
    // ……错误片段直接抛 AiCallException
    JsonNode delta = chunk.path("choices").path(0).path("delta");
    JsonNode piece = delta.path("content");
    if (piece.isTextual()) content.append(piece.asText());
    JsonNode reasoning = delta.path("reasoning_content");
    if (reasoning.isTextual() && !reasoning.asText().isEmpty()) sawReasoning = true;
    if (content.length() > 0 || sawReasoning) gotOutput[0] = true;
}

注意置位的条件是「收到了内容」，不是「收到了一行」。有的上游会先回一个只有 role 的空 delta、或者定时发 : keep-alive 注释行，这些不算开始写。

异常类上加一个标记，不改原来的构造器，免得所有调用方都要跟着动：

public class AiCallException extends RuntimeException {
    private final boolean retryable;
    private final int statusCode;
    /** 上游一个字还没写就断了（读超时前没收到任何输出）：和「忙」一样，时间够就值得再来一遍。 */
    private boolean beforeOutput;

    public boolean isBeforeOutput() { return beforeOutput; }

    AiCallException markBeforeOutput() {
        this.beforeOutput = true;
        return this;
    }
}

markBeforeOutput 是包级可见，只有同包的 client 能打这个标记。连接超时、连接被拒也会走到 ResourceAccessException，这时 gotOutput 一样是 false，也按「没开始写」处理——确实也是，重试是对的。

四、退避要和总时限一起算

当天中午上游连续说忙，老规则「隔 2 秒、最多 3 次」在 7 秒内就把机会用完了，而总时限还剩一百多秒。改成逐次拉长：2、4、8 秒，之后都是 10 秒，最多 5 次。

static final int MAX_ATTEMPTS = 5;
static final long RETRY_WINDOW_MS = 20000L;
static final long BUSY_RETRY_MIN_REMAINING_MS = 95000L;
static final long BUSY_BACKOFF_MS = 2000L;
static final long BUSY_BACKOFF_MAX_MS = 10000L;

/** 第 attempt 次说忙后等多久：2、4、8 秒，之后都是 10 秒。 */
static long busyBackoff(int attempt) {
    return Math.min(BUSY_BACKOFF_MAX_MS, BUSY_BACKOFF_MS << Math.min(10, attempt - 1));
}

Math.min(10, attempt - 1) 是防御性的：左移位数不封顶的话，次数一多 2000L << 63 会溢出成负数，Math.min 就会选中一个负的等待时间。

关键是「还剩多少时间才值得再试」要把这次要等的时间也扣掉：等完 10 秒只剩 90 秒，再试大概率写不完，不如现在就告诉用户失败（失败不扣次数，用户可以过一会儿再点）。

private <T> T withRetry(String system, String user, long totalTimeoutMs, Function<String, T> parse) {
    long started = now();
    long deadline = started + Math.max(1000, totalTimeoutMs);
    AiCallException last = null;
    for (int attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) {
        try {
            return parse.apply(callRaw(system, user, deadline));
        } catch (AiCallException ex) {
            last = ex;
            long now = now();
            long elapsed = now - started;
            if (!ex.isRetryable() || attempt == MAX_ATTEMPTS) break;
            if (isBusy(ex) || ex.isBeforeOutput()) {
                long backoff = busyBackoff(attempt);
                if (deadline - now - backoff < BUSY_RETRY_MIN_REMAINING_MS) {
                    log.warn("文本上游第 {} 次没开始写（忙或排队，已用 {} ms，剩下的时间不够再写一份）: {}", attempt, elapsed, ex.getMessage());
                    break;
                }
                log.warn("文本上游第 {} 次没开始写（忙或排队），{} ms 后再试: {}", attempt, backoff, ex.getMessage());
                pause(backoff);
                continue;
            }
            if (attempt >= 2 || elapsed > RETRY_WINDOW_MS) {
                log.warn("文本上游第 {} 次失败（已用 {} ms，来不及重试）: {}", attempt, elapsed, ex.getMessage());
                break;
            }
            log.warn("文本上游第 {} 次失败，重试一次: {}", attempt, ex.getMessage());
        }
    }
    throw last;
}

几个细节：

• 所有尝试共用同一个 deadline。 重试不是重新计时，callRaw 里每读一行都对同一个 deadline 检查，总耗时不会超过 140 秒。
• 超过总时限抛的异常是 retryable = false。 已经超时了就别再试，第一行 if 直接跳出。
• 「说忙」靠关键字判断。 上游在流里回的错误片段只有一句中文提示，没有结构化的错误码，只能匹配「访问量较大」「稍后重试」「暂时不可用」「overloaded」「unavailable」这些词，外加 HTTP 503。这类判断写成包级 static 方法，方便单测直接调。
• 时间和睡眠都走可覆盖的方法。 now() 和 pause(ms) 是包级可见的普通方法，测试里覆盖掉，后面会用到。

五、怎么测这些分支

MockRestServiceServer 可以绑到同一个 RestTemplate 上，按顺序预设每次请求的响应。难点在于「过了 20 秒」「还剩 95 秒」这类和时间有关的分支，总不能真等。做法是在测试里继承一个匿名子类，覆盖时钟和睡眠：

/** 时钟每读一次往前走 stepMs，用来模拟「已经用了很久」。 */
private ZhiChatClientImpl clientWithClock(final long stepMs) {
    return new ZhiChatClientImpl(restTemplate, objectMapper, properties, new RecipeJsonParser(objectMapper)) {
        private long t = 0;
        @Override
        long now() {
            t += stepMs;
            return t;
        }

        @Override
        void pause(long ms) {
        }
    };
}

没开始写就超时。 ResponseCreator 是个函数式接口，直接在里面抛 SocketTimeoutException，RestTemplate 会把它包成 ResourceAccessException，和线上一模一样：

@Test
void timeoutBeforeAnyOutputIsRetriedLikeBusy() throws Exception {
    // 线上 2026-10-12：上游 30 秒一个字没写，第 34 秒空闲超时；总时限还剩 106 秒，应该再来一遍
    properties.getAi().setTextTimeoutMs(1_000_000);
    ZhiChatClientImpl late = clientWithClock(ZhiChatClientImpl.RETRY_WINDOW_MS + 2_000);
    server.expect(once(), requestTo(URL)).andRespond(request -> {
        throw new SocketTimeoutException("Read timed out");
    });
    server.expect(once(), requestTo(URL)).andRespond(withSuccess(stream(FULL), MediaType.TEXT_EVENT_STREAM));
    assertNotNull(late.generateRecipe(prompt));
    server.verify();
}

时钟每次前进 22 秒，保证「已用时间超过 20 秒窗口」——在老规则下这个测试会失败，新规则下第二次请求成功。

写到一半才超时。 要让响应先吐出一段正文、再在下一次 read 时超时。用 SequenceInputStream 把一段正常的 SSE 和一个「一读就超时」的流接起来，塞进 MockClientHttpResponse：

@Test
void timeoutAfterOutputStartedIsNotRetriedLate() throws Exception {
    properties.getAi().setTextTimeoutMs(1_000_000);
    ZhiChatClientImpl late = clientWithClock(ZhiChatClientImpl.RETRY_WINDOW_MS + 2_000);
    server.expect(once(), requestTo(URL)).andRespond(request -> {
        byte[] head = "data: {\"choices\":[{\"delta\":{\"content\":\"{\\\"title\"}}]}\n\n".getBytes(StandardCharsets.UTF_8);
        InputStream stalled = new InputStream() {
            @Override
            public int read() throws IOException {
                throw new SocketTimeoutException("Read timed out");
            }
        };
        MockClientHttpResponse response = new MockClientHttpResponse(
                new SequenceInputStream(new ByteArrayInputStream(head), stalled), HttpStatus.OK);
        response.getHeaders().setContentType(MediaType.TEXT_EVENT_STREAM);
        return response;
    });
    AiCallException ex = assertThrows(AiCallException.class, () -> late.generateRecipe(prompt));
    assertFalse(ex.isBeforeOutput());
    server.verify();
}

这里只预设了一次请求：如果代码误判成「没开始写」去重试，MockRestServiceServer 会因为收到没预期的第二个请求而报错，所以这个测试同时保证了「不重试」。

退避时间。 纯函数直接断言：

assertEquals(2000L, ZhiChatClientImpl.busyBackoff(1));
assertEquals(4000L, ZhiChatClientImpl.busyBackoff(2));
assertEquals(8000L, ZhiChatClientImpl.busyBackoff(3));
assertEquals(10000L, ZhiChatClientImpl.busyBackoff(4));
assertEquals(10000L, ZhiChatClientImpl.busyBackoff(9));

还踩了一个小坑：原来有个测试「上游连续说忙直到用完次数」用的是真时钟的 client。次数从 3 改成 5、间隔拉长以后，它会真的 Thread.sleep 2+4+8+10 = 24 秒，整个测试套件慢了一截。换成 clientWithClock(1)（时钟几乎不走、pause 不睡）就好了。凡是会睡的代码，睡眠都该能被替换掉。

六、上线后的样子

部署后当天下午上游还在抽风，日志里能看到新规则在工作：

文本上游第 1 次没开始写（忙或排队），2000 ms 后再试: 文本上游流式返回错误: 当前模型访问量较大，请稍后重试
文本上游第 2 次没开始写（忙或排队），4000 ms 后再试: ……
文本上游第 3 次没开始写（忙或排队），8000 ms 后再试: ……
文本上游第 4 次没开始写（忙或排队，已用 32967 ms，剩下的时间不够再写一份）: 文本上游流式返回错误: 服务暂时不可用，请稍后重试

第 4 次之后算上 10 秒等待只剩不到 95 秒，按规则放弃。等上游缓过来，同一时段的冰箱想菜（一次短请求）6 秒就回来了。

要承认的是：重试只能扛住短时间的抖动。上游连续半小时不可用，怎么退避都没用，这时候该做的是让失败便宜一点——失败不扣用户的次数、错误页给一个「再试一次」，而不是在后端死等到 nginx 超时。

七、顺手修的一个解析问题：forty-eight 克

同一天截图时发现，一道香酥鸭的「营养参考」显示成这样：

720 千卡    forty-ei…    fifty-tw…    twelve 克

翻库里的数据：

{"calories":720,"protein":"forty-eight 克","fat":"fifty-two 克","carbs":"twelve 克"}

原因在提示词里，营养字段的格式是这么写的：

- nutrition：{"calories":整数,"protein":"数字 克","fat":"数字 克","carbs":"数字 克"}，每人份；

calories 写的是「整数」，模型老老实实给了 720；三个营养素写的是「数字 克」，模型把「数字」理解成了「用文字写出来的数」。全库只有两道菜中招，但说明这是会复现的。

两处一起改。提示词不再描述类型，直接给示例：

- nutrition：{"calories":整数,"protein":"38克","fat":"20克","carbs":"15克"}（数值仅为格式示例，用阿拉伯数字），每人份；

解析层再兜一道：取第一个阿拉伯数字；没有的话试着把英文数词换成数字；都不行就置空（页面上这一项不显示，总比显示一串英文强）。

private static final Pattern NUMBER = Pattern.compile("\\d+(?:\\.\\d+)?");
private static final List<String> EN_UNITS = Arrays.asList("zero", "one", "two", "three", "four", "five", "six",
        "seven", "eight", "nine", "ten", "eleven", "twelve", "thirteen", "fourteen", "fifteen", "sixteen",
        "seventeen", "eighteen", "nineteen");
private static final List<String> EN_TENS = Arrays.asList("", "", "twenty", "thirty", "forty", "fifty", "sixty",
        "seventy", "eighty", "ninety");

static String grams(String value) {
    if (value == null) return null;
    Matcher m = NUMBER.matcher(value);
    if (m.find()) return m.group() + "克";
    Integer words = englishNumber(value);
    return words == null ? null : words + "克";
}

/** "forty-eight" / "one hundred twenty" → 数字；含不认识的词返回 null。 */
static Integer englishNumber(String value) {
    String v = value.toLowerCase(Locale.ROOT).replace("克", " ").replace('-', ' ').trim();
    if (v.isEmpty()) return null;
    int total = 0;
    boolean any = false;
    for (String word : v.split("\\s+")) {
        if (word.isEmpty() || "and".equals(word) || "g".equals(word) || "gram".equals(word) || "grams".equals(word)) continue;
        int unit = EN_UNITS.indexOf(word);
        int tens = EN_TENS.indexOf(word);
        if (unit >= 0) total += unit;
        else if (tens >= 2) total += tens * 10;
        else if ("hundred".equals(word)) total = Math.max(1, total) * 100;
        else return null;
        any = true;
    }
    return any ? total : null;
}

写第一版时差点埋个雷：想用一个正则 (grams?|g|克|and) 把单位先替换掉，结果 g 会把 eight 里的字母 g 也换掉，变成 ei ht。改成先按空白切词、再整词比较。单测覆盖了这几种写法：

assertEquals("48克", RecipeJsonParser.grams("forty-eight 克"));
assertEquals("52克", RecipeJsonParser.grams("Fifty-two grams"));
assertEquals("120克", RecipeJsonParser.grams("one hundred and twenty g"));
assertEquals("38克", RecipeJsonParser.grams("38 克"));
assertEquals("12.5克", RecipeJsonParser.grams("约12.5g"));
assertEquals(null, RecipeJsonParser.grams("适量"));

线上已有的两条数据用一条带回滚语句的 DML 修正，执行前先 mysqldump 备份了这两行，UPDATE 的 WHERE 里同时带上了旧值，防止误改到别的记录。

八、几点体会

• 超时要分清在量什么。 非流式时读超时是「整份写完的时限」，流式时是「两段之间的空闲时限」，总时限得自己掐，而且要比前面 nginx、客户端的超时都短。
• 同一个异常可能代表完全不同的处境。 Read timed out 既可能是没开始写，也可能是写到一半；能不能重试取决于后者，而这个信息只在读的过程中才有，得自己记下来。
• 进度状态放在调用里，不放在对象上。 单例 client 被多个线程同时用，局部的 boolean[] 比字段安全，也比 ThreadLocal 简单。
• 退避和总时限一起算，等待时间也要扣掉。 剩下的时间不够写完一份，就别再试，早点失败、失败不扣次数。
• 时间和睡眠做成可覆盖的方法。 和时间有关的分支才测得出来，测试也不会真睡。
• 给模型的格式说明用示例，不用类型名。 「数字 克」这种写法对人是格式，对模型可能是内容；解析层永远留一道兜底。