Java 大模型应用 - 流式响应的核心概念与代码实践

大模型生成一段回答往往要几秒甚至更久，等完整结果生成后再一次性返回，用户只能对着空白页面干等。本文介绍流式响应中用到的 Mono、Flux、Sinks、Disposable、flatMap 和背压，并给出一个基于 Spring WebFlux 与 SSE 的完整示例。

为什么大模型应用需要流式响应

传统 HTTP 接口通常在业务处理完成后一次性返回结果。这个方式适合耗时短、结果固定的请求，但不适合大模型逐步生成 Token 的过程。

假设模型生成完整回答需要 10 秒：

• 非流式响应要等待 10 秒，之后一次性显示全部内容。
• 流式响应可以在第一个 Token 产生后立即返回，并持续把后续 Token 推送给前端。

流式响应不会让模型生成得更快，只是让用户更早看到第一个结果。对话界面里的“逐字输出”和“停止生成”，靠的就是持续传输和任务取消。

在基于 Spring WebFlux 的 Java 应用中，一条常见的数据链路如下：

大模型逐步生成 Token
        ↓
Flux<String>
        ↓
Spring WebFlux
        ↓
SSE（text/event-stream）
        ↓
浏览器逐条接收并渲染

准备 Reactor 运行环境

如果项目没有使用 Spring Boot 的依赖管理，可以直接引入 reactor-core。

<dependency>
    <groupId>io.projectreactor</groupId>
    <artifactId>reactor-core</artifactId>
    <version>3.8.6</version>
</dependency>

如果项目已经引入 spring-boot-starter-webflux，Spring Boot 会通过 BOM 管理 Reactor 版本，通常不需要再单独声明 reactor-core 的版本：

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

Mono：表示 0 或 1 个异步结果

Mono<T> 表示一个最多产生一个元素的异步序列，其结果只有三种可能：产生一个值后完成、不产生值直接完成，或者执行失败并发出错误信号。

它适合大模型的非流式调用、数据库单条查询以及只需要完成信号的异步任务。与直接返回 String 相比，返回 Mono<String> 不是说结果已经存在，而是在描述“订阅后如何产生这个结果”。

下面将一个阻塞式模型调用包装成 Mono：

import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

public class MonoChatService {

    // 将阻塞式模型调用包装为单值异步结果
    public Mono<String> chat(String prompt) {
        return Mono.fromCallable(() -> callBlockingModel(prompt))
                .subscribeOn(Schedulers.boundedElastic());
    }

    // 模拟阻塞式大模型客户端
    private String callBlockingModel(String prompt) {
        return "模型对问题的完整回答：" + prompt;
    }
}

这里有两个容易混淆的地方：

1. Mono.fromCallable() 只负责延迟执行并包装结果，不会自动创建后台线程。
2. subscribeOn(Schedulers.boundedElastic()) 才把阻塞任务安排到适合阻塞工作的有界弹性线程池。

在 WebFlux 控制器中应尽量直接返回 Mono，由框架负责订阅，不要为了立即取得结果而调用 block()：

import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Mono;

@RestController
public class MonoChatController {

    // 非流式聊天服务
    private final MonoChatService chatService;

    // 初始化聊天控制器
    public MonoChatController(MonoChatService chatService) {
        this.chatService = chatService;
    }

    // 返回单次完整的大模型回答
    @GetMapping("/api/chat")
    public Mono<String> chat(@RequestParam String prompt) {
        return chatService.chat(prompt);
    }
}

Mono<Void> 只表示任务完成，不携带任何返回值，适合保存会话记录、发送通知这类操作。流式接口的停止接口用的就是它：完成即可，没有业务数据要返回。

Flux：表示 0 到 N 个连续结果

Flux<T> 可以连续产生 0 到 N 个元素。大模型每生成一段文本就发出一个 String，因此 Token 流通常表示为 Flux<String>。

下面用延迟元素模拟模型逐段生成内容：

import java.time.Duration;
import reactor.core.publisher.Flux;

public class FluxChatService {

    // 模拟大模型逐段返回 Token
    public Flux<String> stream(String prompt) {
        return Flux.just("你好，", "正在回答：", prompt)
                .delayElements(Duration.ofMillis(300));
    }
}

Flux 在创建时通常不会立即执行。只有出现订阅者后，数据源才开始产生数据：

// 流式聊天服务
FluxChatService chatService = new FluxChatService();

// 模型返回的 Token 流
Flux<String> tokenStream = chatService.stream("什么是流式响应？");

// 订阅 Token、错误和完成信号
tokenStream.subscribe(
        token -> System.out.print(token),
        error -> System.err.println("生成失败：" + error.getMessage()),
        () -> System.out.println("\n生成完成")
);

一个响应式流不只传递数据，还会传递错误和完成信号：onNext 表示收到一个 Token，onError 表示执行失败并终止，onComplete 表示所有 Token 已经发送完成。

在完整的流式接口中，Flux<String> 会被控制器包装成 SSE 事件流并作为响应返回，浏览器每收到一个事件就渲染一段内容。

Flux.concat 与 concatWith：按顺序拼接多个流

Flux 之间需要首尾相接时，可以使用 Flux.concat 或 concatWith。它们都会等前一个流正常完成后，才订阅下一个流，因此拼接后的元素顺序与各流的先后顺序一致。

Flux.concat 是静态方法，一次接收多个 Publisher：

import reactor.core.publisher.Flux;

public class ConcatExample {

    // 依次输出提示语、模型回答和结束标记
    public Flux<String> assemble() {
        // 首部提示流
        Flux<String> header = Flux.just("开始回答：");

        // 模型回答流
        Flux<String> answer = Flux.just("流式响应", "的核心概念");

        // 尾部结束流
        Flux<String> footer = Flux.just("[END]");

        return Flux.concat(header, answer, footer);
    }
}

concatWith 是实例方法，每次把另一个流拼接到当前流之后，上面的示例可以等价写成：

// 依次拼接提示流、回答流和结束流
Flux<String> combined = header.concatWith(answer).concatWith(footer);

这两种写法等价。

concat 与 flatMap 合并流的方式不同：

方式	订阅时机	元素顺序	总耗时
concat	前一个流完成后才订阅下一个	严格按流的先后顺序	各流耗时之和
flatMap	按并发度同时订阅多个内部流	不保证顺序	接近最慢内部流耗时

大模型应用里，concat 适合顺序明确的拼接，比如在回答流前面加角色提示、结尾追加结束标记；需要并行执行时才用 flatMap。另外 concat 遇到前一个流出错会直接终止，后面的流不会再被订阅。

Sinks：由业务代码主动向流中发送数据

有些数据源本身就返回 Flux，直接把流交给 WebFlux 就行。但智能体系统里，Token 还可能来自回调、消息队列或多个执行阶段，得由业务代码自己往流里写数据，Sinks 就是 Reactor 提供的手动写入入口。

下面创建一个只允许单个订阅者消费的 Token 通道：

import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.util.concurrent.Queues;

public class TokenSink {

    // 单订阅者 Token 通道，最多暂存 256 个尚未处理的数据
    private final Sinks.Many<String> sink =
            Sinks.many().unicast().onBackpressureBuffer(
                    Queues.<String>get(256).get()
            );

    // 向通道发送一个 Token
    public void emit(String token) {
        // 本次发送的结果
        Sinks.EmitResult emitResult = sink.tryEmitNext(token);
        if (emitResult.isFailure()) {
            throw new IllegalStateException("Token 发送失败：" + emitResult);
        }
    }

    // 正常结束 Token 流
    public void complete() {
        sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST);
    }

    // 将可写通道暴露为只读数据流
    public Flux<String> asFlux() {
        return sink.asFlux();
    }
}

三者在数据链路中的职责不同：

角色	作用
Sinks.Many<String>	由业务代码调用 tryEmitNext() 或 emitNext() 写入 Token
Flux<String>	向下游暴露只能订阅的数据流
Subscriber	消费 Token，并接收错误或完成信号

tryEmitNext() 不保证发送成功，它会返回 EmitResult。没有订阅者、流已经终止、并发发送冲突或下游没有需求时，都可能得到失败结果，生产代码不能完全忽略它。

Sinks.many() 还需要根据订阅模型选择具体类型：

• unicast()：只允许一个订阅者，适合一次聊天请求对应一个前端连接。
• multicast()：多个当前订阅者接收后续数据，适合实时广播。
• replay()：新订阅者还可以收到部分或全部历史数据，适合需要回放的场景。

流式接口里，模型 SDK 的回调或事件先写进 Sinks.Many，再由它转发给当前 SSE 请求，SDK 和 Web 层之间的数据传递就靠这一层中转。

Disposable：取消已经建立的订阅

调用 Lambda 形式的 subscribe() 后会得到一个 Disposable。它是本次订阅的取消句柄，可以用来实现“停止生成”。

下面使用会话 ID 保存和取消任务：

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import reactor.core.Disposable;

public class AgentTaskManager {

    // 会话 ID 与流式任务之间的映射
    private final Map<String, Disposable> tasks = new ConcurrentHashMap<>();

    // 注册正在执行的流式任务
    public void register(String conversationId, Disposable task) {
        // 同一会话此前注册的任务
        Disposable previousTask = tasks.put(conversationId, task);
        if (previousTask != null) {
            previousTask.dispose();
        }
    }

    // 停止并移除指定会话的任务
    public boolean stop(String conversationId) {
        // 当前需要停止的任务
        Disposable task = tasks.remove(conversationId);
        if (task == null) {
            return false;
        }
        task.dispose();
        return true;
    }
}

启动模型流并注册任务时，可以这样使用：

// 当前会话的唯一标识
String conversationId = "conversation-1001";

// 模型返回的 Token 流
Flux<String> tokenStream = chatService.stream("介绍 Project Reactor");

// 订阅后得到的取消句柄
Disposable task = tokenStream.subscribe(token -> System.out.print(token));

// 保存任务，供停止接口使用
taskManager.register(conversationId, task);

当用户点击停止按钮时，服务端调用：

// 取消指定会话的订阅
boolean stopped = taskManager.stop("conversation-1001");

dispose() 可以取消 Reactor 订阅并向上游传播取消信号，但“取消订阅”不一定等于“远端大模型已经停止推理”。只有 HTTP 客户端、模型 SDK 和服务端都正确处理取消信号时，底层网络请求和推理任务才会真正结束。如果上游忽略取消，应用只能停止接收结果，不能保证远端计算立即终止。

flatMap：把元素转换为异步任务并合并结果

map 适合一对一的同步转换，例如把 Token 转为大写字符串；如果转换函数返回的仍然是 Mono 或 Flux，继续使用 map 会得到嵌套类型 Flux<Mono<T>>。flatMap 会订阅这些内部异步流，再把结果合并为一条流。

下面并发调用多个异步模型任务：

import java.time.Duration;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class FlatMapExample {

    // 并发处理多个问题，最多同时执行两个内部任务
    public Flux<String> answerAll() {
        // 等待处理的问题流
        Flux<String> prompts = Flux.just("问题一", "问题二", "问题三");

        // 合并各个异步模型调用的结果
        Flux<String> answers = prompts.flatMap(this::callModel, 2);
        return answers;
    }

    // 模拟一次异步模型调用
    private Mono<String> callModel(String prompt) {
        return Mono.just("回答：" + prompt)
                .delayElement(Duration.ofMillis(300));
    }
}

第二个参数 2 是并发度，表示最多同时订阅两个内部任务。限制并发度可以避免一次性向模型服务或外部工具发起过多请求。

flatMap 不保证结果与输入顺序一致。如果业务依赖原始顺序，可以选择 concatMap 按顺序逐个执行，或者选择 flatMapSequential 并发启动后按原输入顺序发送结果。

在完整智能体中，一次回答往往需要同时调用检索、代码执行等多个异步工具，flatMap 可以让这些子任务并发执行，再合并各工具的返回结果继续生成。

Backpressure：让生产速度服从消费能力

背压解决的是生产者速度超过消费者处理能力的问题。响应式流中的下游会通过 Subscription.request(n) 表达自己还能接收多少个元素，上游根据需求发送数据，而不是不受限制地持续推送。

下面的订阅者每次只请求两个元素，处理完一批后再请求下一批：

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;

public class BackpressureExample {

    // 演示下游分批请求数据
    public void consumeInBatches() {
        // 等待消费的数据流
        Flux<Integer> source = Flux.range(1, 10);

        source.subscribe(new BaseSubscriber<Integer>() {

            // 当前批次已经处理的数量
            private int consumedInBatch;

            // 建立订阅后请求第一批数据
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                request(2);
            }

            // 消费一个元素，并在一批完成后继续请求
            @Override
            protected void hookOnNext(Integer value) {
                System.out.println("消费数据：" + value);
                consumedInBatch++;
                if (consumedInBatch == 2) {
                    consumedInBatch = 0;
                    request(2);
                }
            }
        });
    }
}

并非所有数据源都能减慢生产速度。例如外部回调可能不关心下游需求，仍然持续发送数据。此时需要明确选择溢出策略：

// 无法根据下游需求减速的上游数据流
Flux<String> fastSource = getFastTokenSource();

// 缓存最多 100 个暂时来不及处理的 Token
Flux<String> buffered = fastSource.onBackpressureBuffer(100);

// 丢弃下游暂时无法接收的 Token
Flux<String> dropped = fastSource.onBackpressureDrop(
        token -> System.out.println("丢弃 Token：" + token)
);

// 消费跟不上时只保留最新的 Token
Flux<String> latest = fastSource.onBackpressureLatest();

常见策略的取舍如下：

策略	适用情况	风险
Buffer	每个元素都不能丢失，并且流量峰值可控	无界缓冲可能持续占用内存；有界缓冲需要处理溢出
Drop	允许舍弃部分高频状态或监控数据	被丢弃的数据无法恢复
Latest	只关心最新状态，例如进度百分比	不适合正文 Token，会导致回答缺字
Error	消费跟不上时立即失败	需要上层提供重试或错误提示

大模型回答中的正文 Token 通常不能丢弃，因此更常使用有界 Buffer，同时限制模型调用并发度、监控队列长度并处理客户端断开。onBackpressureBuffer() 的无参形式使用无界缓冲，不能因为写法简单就默认适合生产环境。

使用 Spring WebFlux 与 SSE 实现完整流式接口

SSE 使用普通 HTTP 长连接，并通过 text/event-stream 持续发送事件，适合服务端单向推送大模型 Token。下面把前面的概念组合成一个可以停止生成的流式聊天服务。

首先实现任务管理器和模型流桥接服务：

import java.time.Duration;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.springframework.stereotype.Service;
import reactor.core.Disposable;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Sinks;
import reactor.util.concurrent.Queues;

@Service
public class StreamingChatService {

    // 会话与完整流式任务之间的映射
    private final Map<String, ChatTask> tasks = new ConcurrentHashMap<>();

    // 启动一次模型生成并返回 Token 流
    public Flux<String> stream(String conversationId, String prompt) {
        // 当前请求独占的有界 Token 通道
        Sinks.Many<String> sink =
                Sinks.many().unicast().onBackpressureBuffer(
                        Queues.<String>get(256).get()
                );

        // 模型 SDK 返回的原始 Token 流
        Flux<String> modelStream = callModelStream(prompt);

        // 将模型事件转发到当前请求的 Sink
        Disposable sourceTask = modelStream.subscribe(
                token -> sink.emitNext(token, Sinks.EmitFailureHandler.FAIL_FAST),
                error -> sink.emitError(error, Sinks.EmitFailureHandler.FAIL_FAST),
                () -> sink.emitComplete(Sinks.EmitFailureHandler.FAIL_FAST)
        );

        // 当前请求对应的完整流式任务
        ChatTask currentTask = new ChatTask(sourceTask, sink);

        // 同一会话此前尚未结束的任务
        ChatTask previousTask = tasks.put(conversationId, currentTask);
        if (previousTask != null) {
            previousTask.stop();
        }

        return sink.asFlux()
                .doFinally(signalType -> {
                    tasks.remove(conversationId, currentTask);
                    currentTask.disposeSource();
                });
    }

    // 停止指定会话的模型生成任务
    public boolean stop(String conversationId) {
        // 当前需要取消的完整流式任务
        ChatTask task = tasks.remove(conversationId);
        if (task == null) {
            return false;
        }
        task.stop();
        return true;
    }

    // 模拟模型 SDK 的流式输出，实际项目中替换为真实调用
    private Flux<String> callModelStream(String prompt) {
        return Flux.just("收到问题：", prompt, "，正在生成回答。")
                .delayElements(Duration.ofMillis(300));
    }

    // 同时管理上游订阅和下游 Token 通道
    private static class ChatTask {

        // 模型流的取消句柄
        private final Disposable sourceTask;

        // 向 SSE 客户端发送数据的 Token 通道
        private final Sinks.Many<String> sink;

        // 初始化完整流式任务
        private ChatTask(Disposable sourceTask, Sinks.Many<String> sink) {
            this.sourceTask = sourceTask;
            this.sink = sink;
        }

        // 取消上游订阅并结束下游 Token 流
        private void stop() {
            sourceTask.dispose();
            sink.tryEmitComplete();
        }

        // 仅释放上游订阅
        private void disposeSource() {
            sourceTask.dispose();
        }
    }
}

doFinally() 会在流正常完成、发生错误或被取消后执行，因此浏览器断开 SSE 连接时也能清理任务。这里的 Sinks.Many 只服务于一次请求，使用 unicast() 与单个 SSE 客户端对应。

然后提供 SSE 接口和停止接口：

import org.springframework.http.MediaType;
import org.springframework.http.codec.ServerSentEvent;
import org.springframework.web.bind.annotation.DeleteMapping;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@RestController
public class StreamingChatController {

    // 流式聊天服务
    private final StreamingChatService chatService;

    // 初始化流式聊天控制器
    public StreamingChatController(StreamingChatService chatService) {
        this.chatService = chatService;
    }

    // 以 SSE 形式持续返回模型 Token
    @GetMapping(
            value = "/api/conversations/{conversationId}/stream",
            produces = MediaType.TEXT_EVENT_STREAM_VALUE
    )
    public Flux<ServerSentEvent<String>> stream(
            @PathVariable String conversationId,
            @RequestParam String prompt
    ) {
        return chatService.stream(conversationId, prompt)
                .map(token -> ServerSentEvent.builder(token)
                        .event("token")
                        .build());
    }

    // 停止指定会话正在执行的生成任务
    @DeleteMapping("/api/conversations/{conversationId}/stream")
    public Mono<Void> stop(@PathVariable String conversationId) {
        chatService.stop(conversationId);
        return Mono.empty();
    }
}

浏览器可以使用 EventSource 接收 SSE：

// 当前会话的唯一标识
const conversationId = "conversation-1001";

// 经过 URL 编码的用户问题
const prompt = encodeURIComponent("什么是流式响应？");

// 与服务端建立 SSE 长连接
const eventSource = new EventSource(
  `/api/conversations/${conversationId}/stream?prompt=${prompt}`
);

// 将收到的 Token 追加到页面
eventSource.addEventListener("token", (event) => {
  document.querySelector("#answer").textContent += event.data;
});

// 连接异常时关闭客户端连接
eventSource.onerror = () => {
  eventSource.close();
};

// 请求服务端停止生成并关闭 SSE 连接
async function stopGeneration() {
  await fetch(`/api/conversations/${conversationId}/stream`, {
    method: "DELETE"
  });
  eventSource.close();
}

这条链路中各个核心概念分别承担下面的职责：

概念	在示例中的作用
Mono	停止接口只表示异步完成，不返回业务数据
Flux	表示模型连续产生的 Token 序列，并作为 SSE 响应返回
Sinks	把模型 SDK 的事件主动转发给当前 SSE 请求
Disposable	保存模型流订阅，用于停止生成和断开后的资源清理
flatMap	完整智能体中可用于并发执行多个异步工具或子任务
Backpressure	协调 Token 生产与网络发送速度，防止数据无限积压