使用 Java 21 和 Spring Boot 从 LLM API 流式响应

1. 概览

一个来自 LLM 的长答案可能需要 10–30 秒才能完全生成。如果你等到整个答案生成完毕才返回，用户会看到一个空白的屏幕。流式传输允许你在模型生成每个 token 时立即接收，这样第一个字符通常会在不到一秒内出现。

本文将指导你使用 java.net.http.HttpClient 从兼容 OpenAI Chat Completions 的 API 读取流，然后使用 Spring WebClient。接下来，你将通过 Server-Sent Events (SSE) 将流推送到浏览器，并为整个流程编写测试。

2. LLM API 的流格式

当使用 "stream": true 发送请求时，API 不会返回单个 JSON。相反，它返回 Content-Type: text/event-stream，每个事件都是一行 data:：

data: {"id":"c1","choices":[{"index":0,"delta":{"role":"assistant","content":""}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{"content":"Xin"}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{"content":" chào"}}]}

data: {"id":"c1","choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}

data: [DONE]

你需要记住三点：

• 文本内容位于 choices[0].delta.content。某些 chunk 没有这个字段，例如第一个 chunk 只有 role，或最后一个 chunk 只有 finish_reason。
• 事件之间由空行分隔。
• data: [DONE] 表示流已结束。这个值不是 JSON，所以你不应该将其传递给 Jackson。

3. 将解析逻辑分离到单独的类

解析逻辑应该在独立的类中，以便于测试。这个类只需要 Jackson，它已经包含在所有 Spring Boot starter web 中：

package dev.nonla.chat;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;

import java.util.Optional;

public class ChatChunkParser {

    public static final String DONE = "[DONE]";
    private final ObjectMapper mapper = new ObjectMapper();

    /** 接收完整的一行 SSE，例如：data: {...} */
    public Optional<String> parseLine(String line) {
        if (!line.startsWith("data:")) {
            return Optional.empty();
        }
        return parsePayload(line.substring(5).trim());
    }

    /** 接收 "data:" 后的 payload 部分 */
    public Optional<String> parsePayload(String payload) {
        if (payload.isEmpty() || DONE.equals(payload)) {
            return Optional.empty();
        }
        try {
            JsonNode content = mapper.readTree(payload)
                    .path("choices").path(0).path("delta").path("content");
            return content.isTextual() && !content.asText().isEmpty()
                    ? Optional.of(content.asText())
                    : Optional.empty();
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("无效的 chunk：" + payload, e);
        }
    }
}

这个类的工作方式如下：

• parseLine 忽略空行以及 event: 或 : keep-alive 之类的行。
• Jackson 的 path(...) 在节点缺失时不会抛出异常。如果 chunk 没有 content，结果是 Optional.empty()。
• [DONE] 在调用 readTree 之前被拦截，所以不会导致解析错误。

4. 使用 java.net.http.HttpClient 进行流式传输

如果你只需要一个 CLI 或简单的 job，JDK 中内置的 HttpClient 就足够了。BodyHandlers.ofLines() 返回一个 Stream<String>，它根据从网络接收的数据逐步读取，而不是等待下载完整的 body：

package dev.nonla.chat;

import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.net.http.HttpResponse.BodyHandlers;
import java.time.Duration;
import java.util.Optional;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class StreamingChatDemo {

    public static void main(String[] args) throws Exception {
        String baseUrl = System.getenv("LLM_BASE_URL");
        String apiKey = System.getenv("LLM_API_KEY");

        String body = """
            {"model": "gpt-4o-mini", "stream": true,
             "messages": [{"role": "user",
                           "content": "用三句话解释 Order 和 Customer 之间的关系"}]}
            """;

        HttpRequest request = HttpRequest.newBuilder(URI.create(baseUrl + "/v1/chat/completions"))
                .header("Authorization", "Bearer " + apiKey)
                .header("Content-Type", "application/json")
                .header("Accept", "text/event-stream")
                .timeout(Duration.ofSeconds(60))
                .POST(HttpRequest.BodyPublishers.ofString(body))
                .build();

        HttpClient client = HttpClient.newHttpClient();
        HttpResponse<Stream<String>> response = client.send(request, BodyHandlers.ofLines());

        if (response.statusCode() != 200) {
            String error = response.body().collect(Collectors.joining("\n"));
            throw new IllegalStateException("HTTP " + response.statusCode() + ": " + error);
        }

        ChatChunkParser parser = new ChatChunkParser();
        try (Stream<String> lines = response.body()) {
            lines.takeWhile(line -> !line.equals("data: [DONE]"))
                 .map(parser::parseLine)
                 .flatMap(Optional::stream)
                 .forEach(token -> {
                     System.out.print(token);
                     System.out.flush();
                 });
        }
        System.out.println();
    }
}

主要步骤的解释：

• 必须先检查状态码。当 API 返回错误（401, 429…）时，body 是 JSON 错误而不是 SSE。
• takeWhile 在遇到 [DONE] 时停止流。try-with-resources 块关闭流，连接也随之释放。
• System.out.flush() 帮助文本立即显示。没有这一行，终端可能会缓冲输出然后一次性打印。

结果是文本在终端上逐渐显示：

一个 Customer 可以下达多个 Order，而每个 Order 只属于一个 Customer...

注意： HttpRequest 的 timeout(...) 仅适用于接收响应头之前。它不限制读取 body 的时间。如果模型在中途“卡住”，循环将永远等待。这是改用第 5 节 WebClient 的一个原因。

5. 使用 Spring WebClient 进行流式传输

在 Spring Boot 服务中，WebClient 给你一个 Flux<String> 并支持背压、超时和取消。你需要添加依赖：

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-webflux</artifactId>
</dependency>

连接配置在 application.properties 中设置：

llm.base-url=${LLM_BASE_URL}
llm.api-key=${LLM_API_KEY}
llm.model=gpt-4o-mini

接下来，声明 WebClient bean：

@Configuration
class LlmClientConfig {

    @Bean
    WebClient llmWebClient(@Value("${llm.base-url}") String baseUrl,
                           @Value("${llm.api-key}") String apiKey) {
        return WebClient.builder()
                .baseUrl(baseUrl)
                .defaultHeader(HttpHeaders.AUTHORIZATION, "Bearer " + apiKey)
                .build();
    }
}

Service 调用 API 并返回包含 token 的 Flux：

@Service
public class ChatStreamService {

    private static final Logger log = LoggerFactory.getLogger(ChatStreamService.class);
    private static final ParameterizedTypeReference<ServerSentEvent<String>> SSE_TYPE =
            new ParameterizedTypeReference<>() {};

    private final WebClient llmWebClient;
    private final String model;
    private final ChatChunkParser parser = new ChatChunkParser();

    public ChatStreamService(WebClient llmWebClient, @Value("${llm.model}") String model) {
        this.llmWebClient = llmWebClient;
        this.model = model;
    }

    public Flux<String> stream(String question) {
        Map<String, Object> body = Map.of(
                "model", model,
                "stream", true,
                "messages", List.of(Map.of("role", "user", "content", question)));

        return llmWebClient.post()
                .uri("/v1/chat/completions")
                .contentType(MediaType.APPLICATION_JSON)
                .accept(MediaType.TEXT_EVENT_STREAM)
                .bodyValue(body)
                .retrieve()
                .bodyToFlux(SSE_TYPE)
                .mapNotNull(ServerSentEvent::data)
                .takeWhile(data -> !ChatChunkParser.DONE.equals(data))
                .mapNotNull(data -> parser.parsePayload(data).orElse(null))
                .timeout(Duration.ofSeconds(30))
                .doOnCancel(() -> log.info("客户端已取消问题的流：{}", question));
    }
}

处理链中的每一步：

• bodyToFlux(SSE_TYPE) 让 Spring 自动按空行分隔事件并移除 data: 前缀。你不需要自己处理缓冲。
• takeWhile 在遇到 [DONE] 时完成 Flux。
• timeout(Duration.ofSeconds(30)) 应用于连续两个元素之间的间隔，包括第一个元素。如果 30 秒内没有新 token，Flux 报告 TimeoutException。这是第 4 节 HttpClient 所缺少的。
• 当 API 返回 4xx/5xx 时，retrieve() 在有任何 token 之前抛出 WebClientResponseException。

5.1. 在流完成后保存完整响应

你通常需要保存完整的响应以用于记录历史或计算成本。StringBuilder 必须为每个 subscription 单独创建。因此，你将处理链包装在 Flux.defer 中：

public Flux<String> streamAndRecord(String question) {
    return Flux.defer(() -> {
        StringBuilder full = new StringBuilder();
        return stream(question)
                .doOnNext(full::append)
                .doOnComplete(() -> log.info("完整响应（{} 字符）：{}",
                        full.length(), full));
    });
}

如果 StringBuilder 在 defer 外部声明，两个并发请求会写入同一缓冲区。这个错误在手动测试时很难发现，因为你通常一次只发送一个请求。

6. 将流推送到浏览器

Controller 返回 Flux<ServerSentEvent<...>>，其中 produces = text/event-stream：

public record ChatChunk(String text) {}

@RestController
class ChatController {

    private final ChatStreamService chatStreamService;

    ChatController(ChatStreamService chatStreamService) {
        this.chatStreamService = chatStreamService;
    }

    @GetMapping(path = "/api/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
    Flux<ServerSentEvent<ChatChunk>> stream(@RequestParam("q") String question) {
        return chatStreamService.streamAndRecord(question)
                .map(token -> ServerSentEvent.builder(new ChatChunk(token)).event("token").build())
                .concatWith(Mono.just(ServerSentEvent.builder(new ChatChunk("")).event("done").build()))
                .onErrorResume(ex -> Mono.just(ServerSentEvent
                        .builder(new ChatChunk("模型正忙，请稍后重试。"))
                        .event("error").build()));
    }
}

你可以使用 curl -N 测试端点。-N 标志禁用 curl 的缓冲：

curl -N "http://localhost:8080/api/chat/stream?q=Order%20la%20gi"

event:token
data:{"text":"Order"}

event:token
data:{"text":" là"}

event:token
data:{"text":" đơn hàng"}

event:done
data:{"text":""}

有三个值得注意的设计决策：

• Token 包装在 JSON (ChatChunk) 中，不是发送原始字符串。 根据 SSE 规范，接收端将删除 data: 后面的一个空格。原始发送的 token " là" 在浏览器中会变成 "là"，单词会粘在一起。包含 \n 的 token 也会被分成多个 data: 行。将其包装在 JSON 中有助于避免两个问题。
• 错误作为 error 事件发送。 一旦第一个字节已发送，状态 200 已在头中。你无法再更改为 500。客户端必须监听 error 事件。
• done 事件被显式发送。 浏览器的 EventSource 在流关闭时会自动重新连接。客户端需要 done 事件来知道响应已完成并调用 close()。

当用户关闭标签页时，WebFlux 取消 subscription。取消信号反向传播到 WebClient，Reactor Netty 关闭到 LLM API 的连接，提供商停止生成 token。这样你就不会为没有人阅读的响应部分付费。

7. 编写测试

7.1. 测试 parser

class ChatChunkParserTest {

    private final ChatChunkParser parser = new ChatChunkParser();

    @Test
    void givenContentChunk_whenParseLine_thenReturnsToken() {
        String line = "data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\" chào\"}}]}";

        assertThat(parser.parseLine(line)).contains(" chào");
    }

    @Test
    void givenDoneOrRoleOnlyChunk_whenParseLine_thenReturnsEmpty() {
        assertThat(parser.parseLine("data: [DONE]")).isEmpty();
        assertThat(parser.parseLine(
                "data: {\"choices\":[{\"delta\":{\"role\":\"assistant\"}}]}")).isEmpty();
        assertThat(parser.parseLine(": keep-alive")).isEmpty();
    }

    @Test
    void givenBrokenJson_whenParseLine_thenThrows() {
        assertThatThrownBy(() -> parser.parseLine("data: {oops"))
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("无效的 chunk");
    }
}

第一个测试确认 token 开头的空格被保留。这是整个流程中最容易丢失的细节。

7.2. 使用 WebTestClient 进行 controller 测试

你 mock ChatStreamService 以进行不调用真实 API 的测试。@MockitoBean 来自 Spring Boot 3.4。对于较早的版本，使用 @MockBean。

@WebFluxTest(ChatController.class)
class ChatControllerTest {

    @Autowired
    WebTestClient webTestClient;

    @MockitoBean
    ChatStreamService chatStreamService;

    @Test
    void whenStreaming_thenSendsTokensThenDone() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.just("Xin", " chào", "!"));

        List<ServerSentEvent<String>> events = webTestClient.get()
                .uri("/api/chat/stream?q=hi")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .exchange()
                .expectStatus().isOk()
                .returnResult(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
                .getResponseBody()
                .collectList()
                .block(Duration.ofSeconds(5));

        assertThat(events).extracting(ServerSentEvent::event)
                .containsExactly("token", "token", "token", "done");
        assertThat(events.get(1).data()).isEqualTo("{\"text\":\" chào\"}");
    }

    @Test
    void whenUpstreamFails_thenSendsErrorEvent() {
        when(chatStreamService.streamAndRecord("hi"))
                .thenReturn(Flux.concat(Flux.just("Xin"),
                        Flux.error(new TimeoutException())));

        List<ServerSentEvent<String>> events = webTestClient.get()
                .uri("/api/chat/stream?q=hi")
                .accept(MediaType.TEXT_EVENT_STREAM)
                .exchange()
                .returnResult(new ParameterizedTypeReference<ServerSentEvent<String>>() {})
                .getResponseBody()
                .collectList()
                .block(Duration.ofSeconds(5));

        assertThat(events).extracting(ServerSentEvent::event)
                .containsExactly("token", "error");
    }
}

第二个测试模拟在中途发生错误的情况。客户端接收到已生成的 token，然后是 error 事件，响应仍然带有状态 200。

8. 常见错误和何时不应该使用流

• 使用 bodyToMono(String.class)。 这种方法会将整个 body 聚合在一起然后返回，所以你失去了流式传输的所有好处。
• 反向代理缓冲响应。 Nginx 默认缓冲响应，客户端会在最后同时接收所有 token。你需要为这个路由禁用 proxy_buffering，或添加头 X-Accel-Buffering: no。代理层的 gzip 压缩也可能导致缓冲。
• 在 Spring MVC controller 中使用 .block()。 如果你的项目使用 spring-boot-starter-web（Servlet），你仍然可以返回 Flux 或使用 SseEmitter。不要 block() 然后逐部分写入，因为每个请求会在整个 30 秒内占用线程池中的一个线程。
• 在接收不完整时显示结构化输出。 如果你要求模型返回 JSON 以便解析或调用工具，不完整的 chunk 无法解析。此时你仍然可以进行流式传输，但必须等到接收完整后才能验证。如果不需要逐步显示，通常请求（不使用流式传输）会更简单。
• 计算 token 使用量。 默认情况下，流式传输时的 chunk 没有 usage 字段。许多 API 支持选项，如 "stream_options": {"include_usage": true} 来发送额外的最后一个 chunk，其中包含 usage 和空的 choices。第 3 节的 parser 安全地忽略这个 chunk，因为 path(0) 不存在。

注意： 一旦已向用户发送部分 token，就不应该自动重试。重试会生成不同的响应，拼接到旧部分会形成混乱的内容。只在错误发生在第一个 token 之前时重试，例如从 retrieve() 收到 429 或 503。

9. 结论

从 LLM API 流式传输响应本质上是读取 Server-Sent Events。你读取每一行 data:，获取 choices[0].delta.content，并在遇到 [DONE] 时停止。JDK 的 HttpClient 对于小型工具就足够了。在 Spring Boot 服务中，WebClient 结合 Flux 为你提供 token 之间的超时和向上游进行取消的能力。

当将流推送回浏览器时，你应该将 token 包装在 JSON 中以保留空格和换行符。你还应该显式发送 done 和 error 事件，因为状态码从第一个字节开始就已固定。最后，检查反向代理以确保它不会默默地缓冲整个响应。