SSE 父子流与流编排

一句话：把多个模型或分片的输出合并成一条 SSE，难点不是 merge 这个 API，而是子流失败要隔离、取消要沿父子链传到每个上游、慢客户端下预取不能无限放大。

适用 Reactor 3.6.x / Spring Boot 3.x / JDK 17+。SSE 帧格式、SseEmitter 与 WebFlux 基础写法见 14-Java流式输出与SSE；本文接着讲多路编排。

为什么需要

一个回答可能同时调用多个模型、多个分片或多次工具，每个来源是一条独立数据流，前端却只想要一条按 token 到达的 SSE。逐个请求再拼接会把总延迟变成所有来源之和；嵌套 subscribe 又会让取消、错误和顺序失控。父子流要解决的就是并发、合并和取消三件事：分别对应合并操作符、事件标识和取消链。

核心机制

父子流、冷热流与 HTTP/2 stream

• 父流/子流：父流产生子任务（如 Flux.fromIterable(models)），子流是 flatMap/concatMap/flatMapSequential 返回的 Publisher，通常是一次模型调用或分片。
• 冷流/热流：冷流每次订阅重新执行，适合每个 SSE 请求重新调模型；cache()、share()、Sinks.Many 是热流，多订阅共享数据，重订阅可能不请求，也可能重复计费。
• SSE 与 HTTP/2 stream：前者是一条 HTTP 响应里的应用层事件序列，后者是传输层帧复用单位，生命周期、取消和背压不是同一套概念。

父流取消会向下传播到所有子订阅；父流完成则要等所有子流结束。每个子流有独立订阅，不能假设子流之间共享状态；需要共享资源时用 usingWhen 或显式 Disposable 管理。取消只在子流本身可取消时有效——WebClient 响应流可以，自己写的阻塞循环不行。

合并语义

操作符	并发	输出顺序	慢子流的影响	适用
concatMap	1	严格按源顺序	一个慢子流拖住后面全部	顺序重要、下游容量小
flatMap	默认 256，可设 concurrency	交错，不保证顺序	慢子流只占一个并发槽	多模型抢延迟，顺序不重要
flatMapSequential	默认 256	按源顺序输出	慢子流阻塞后续结果出队，其他子流仍执行	要顺序又要并发，能接受缓冲
Flux.merge	固定子流全部订阅	交错	同 flatMap，但子流数量在组装时确定	子流少且固定

选型判据是顺序是否属于业务语义：多个模型分栏展示，顺序不重要，用 flatMap；多个分片拼成一段有序文本，用 concatMap 或 flatMapSequential；子流数量固定且很少，Flux.merge 最直接。

flatMap 的 concurrency=256、prefetch=256 是 Reactor 3.6 的默认值。对 LLM 来说，256 个在途请求很可能超过配额；显式写成 flatMap(this::call, 4, 16)，让并发和每个子流的预取都有上限。预取不是免费缓存，慢客户端会让这些队列堆在内存里。

最小上手

两个模型并发输出，最后只发一次 done：

import reactor.core.publisher.Flux;
import java.time.Duration;
record Token(String source, int id, String data, String event) {
    static Token done() { return new Token("all", -1, "[DONE]", "done"); }
}
public class MergeDemo {
    static Flux<Token> model(String s, int n) {
        return Flux.range(1, n).delayElements(Duration.ofMillis(200))
                .map(i -> new Token(s, i, s + "-" + i, "token"))
                .onErrorResume(e -> Flux.just(new Token(s, -1, "error", "error")));
    }
    static Flux<Token> merged() { return Flux.merge(model("a", 3), model("b", 3)).concatWith(Flux.just(Token.done())); }
    public static void main(String[] args) { merged().doOnNext(t -> System.out.println(t.source() + "/" + t.id() + " " + t.data())).blockLast(); }
}

子流内部先 onErrorResume 成一条 event: error，所以一个模型失败不会让 Flux.merge 整体失败，另一个模型还能继续。concatWith(done) 放在 merge 之后，确保 done 不会抢在子流完成前发出；id 用 source-id 保持全局可区分。

生产动作

MVC：取消链与串行 send

客户端断开、超时或 send 抛 IOException 时，必须取消父流，再沿 merge/flatMap 取消所有子流，最后关闭 LLM 连接：

SseEmitter emitter = new SseEmitter(300_000L);
AtomicBoolean closed = new AtomicBoolean(); AtomicReference<Disposable> ref = new AtomicReference<>();
ScheduledExecutorService hb = Executors.newSingleThreadScheduledExecutor();
Runnable cancel = () -> {
    if (!closed.compareAndSet(false, true)) return;
    hb.shutdownNow(); if (ref.get() != null) ref.get().dispose();
};
emitter.onCompletion(cancel); emitter.onError(e -> cancel.run());
emitter.onTimeout(() -> { cancel.run(); synchronized (emitter) { emitter.complete(); } });
hb.scheduleAtFixedRate(() -> { try { sendComment(emitter, closed); } catch (IOException e) { cancel.run(); } }, 15, 15, TimeUnit.SECONDS);
ref.set(merged().subscribe(
        t -> send(emitter, t, closed),
        e -> { send(emitter, new Token("all", -1, String.valueOf(e.getMessage()), "error"), closed);
               complete(emitter, closed); },
        () -> complete(emitter, closed)));
if (closed.get()) ref.get().dispose(); // 订阅建立前已断开
return emitter;

心跳线程和 token 线程不能同时写同一个 SseEmitter，必须共用一把锁或一个发送队列：

static void send(SseEmitter e, Token t, AtomicBoolean closed) {
    if (closed.get()) return;
    try { synchronized (e) { e.send(SseEmitter.event().id(t.source() + "-" + t.id())
            .event(t.event()).data(t.data())); } }
    catch (IOException ex) { if (closed.compareAndSet(false, true)) synchronized (e) { e.completeWithError(ex); } }
}
static void sendComment(SseEmitter e, AtomicBoolean closed) throws IOException { if (!closed.get()) synchronized (e) { e.send(SseEmitter.event().comment("ping")); } }
static void complete(SseEmitter e, AtomicBoolean closed) { if (closed.compareAndSet(false, true)) synchronized (e) { e.complete(); } }

closed 阻止完成后的重复写入和重复 complete()，清理动作保持幂等：dispose() 和 shutdownNow() 可以重复调用。若 ref 赋值前客户端已断开，订阅返回后还要再检查 closed 并 dispose()。客户端半开连接最终靠下一次心跳或 send 抛 IOException 发现，所以心跳不能省。

WebFlux：取消随订阅传播

return merged()
        .map(t -> ServerSentEvent.<String>builder().id(t.source() + "-" + t.id())
                .event(t.event()).data(t.data()).build())
        .timeout(Duration.ofSeconds(10)).doOnCancel(() -> log.info("client cancelled"))
        .doFinally(signal -> closeUpstreams())
        .onErrorResume(e -> Flux.just(ServerSentEvent.<String>builder()
                .event("error").data("stream failed").build()));

框架在客户端断开时取消返回的 Flux，取消沿 merge 传到每个子流。需要「每个子流一个连接、结束必关」时用 Flux.usingWhen；doFinally 适合记录终态和释放幂等资源，但不能保证异步资源一定关闭。

背压与慢客户端

SSE 没有浏览器侧的 request(n)，浏览器不会告诉服务端「我显示不过来了」。应用层必须自己限制：

• flatMap 的 concurrency/prefetch 直接放大在途请求数；多模型场景从 4/16 这类小值开始，按下游配额调。
• MVC 的 send 是同步写，慢客户端会让发送线程和容器缓冲区堆积；用有界队列加单发送线程，队列满时发 event: error 并关闭，不要用无界 onBackpressureBuffer()。
• MVC 用 SseEmitter 的超时控制整条连接（示例 5 分钟）；WebFlux 的 timeout 约束相邻事件间隔，总时长要另设 deadline 检查。超过阈值就断开并记录。
• WebFlux 的响应写回可以对 socket 形成传输层背压，但 limitRate 只控制请求节奏，不会让浏览器变快。
• 监控活跃 SSE 数、每个子流在途数、发送队列长度、send 耗时、取消数和子流错误数；队列持续上涨先断慢客户端，不要加内存。

错误、完成与资源

子流错误在子流内 onErrorResume，父流继续；父流错误在末端转成一条错误事件再结束。部分成功时 done 事件带上成功/失败计数，不要只发 [DONE] 让客户端猜。AtomicBoolean 或 Disposable.Swap 防重复完成；usingWhen 把创建、取消、完成、错误四种终止信号都映射到清理动作。

边界

• 嵌套 subscribe 不会继承父订阅：parent.subscribe(x -> child.subscribe(...)) 里，父流取消不会取消子流，多个子流还会并发写同一个 emitter，顺序和资源都失控。
• 不要 blockLast() 或阻塞请求线程：下面这种写法会把异步流重新变成同步等待，客户端断开后没有 Disposable 可取消，异常还被吞掉。

public SseEmitter bad() {
    SseEmitter emitter = new SseEmitter(0L);
    merged().doOnNext(t -> { try { emitter.send(SseEmitter.event().data(t.data())); } catch (Exception ignored) {} })
            .blockLast(); // 错：占住请求线程，取消链没接上
    return emitter;
}

• 取消能力取决于子流：阻塞循环、不可中断的 SDK 调用不会因为 dispose() 立刻停止；要选支持取消的客户端，或在边界处检查中断标记。
• done 的语义要明确：全部成功、部分成功、全部失败都应可区分；不要在第一个子流完成时就发 [DONE]。
• 有界缓冲不是可选项：慢客户端、下游限流和子流预取叠加时，无界队列只会把故障推迟成 OOM。

讲清楚

多路 SSE 的本质是父流把多个子流组织起来，再把事件合并到一条 HTTP 响应。flatMap 抢延迟但不保序，concatMap 保序但串行，flatMapSequential 在两者之间；并发和预取必须显式设置，否则 256 的默认值会放大在途请求。

生产上最重要的是生命周期：客户端断开 → 框架取消父流 → dispose() 传到每个子流 → 关闭 LLM 连接并停止计费。子流错误各自降级，done 等所有子流结束再发；MVC 要串行化 send 并用心跳发现半开连接，WebFlux 依赖可取消的响应流。嵌套 subscribe、blockLast() 和无界缓冲都会破坏这条链。

适用 Reactor 3.6.x / Spring Boot 3.x / JDK 17+ | 初稿 2026-10