ASP.NET Core SSE 流式接口实战：IAsyncEnumerable + ResponseHeadersRead + 上游阶段式流解析

大多数业务接口都是「一问一答」：请求进去，等几十秒，结果一次性返回。只要上游慢一点，用户就只能盯着转圈，前端也拿不到任何中间状态。SSE（Server-Sent Events）就是把这堵墙拆开的方式——服务端保持 HTTP 连接不关，把过程按事件一条条推给浏览器。

本文基于 SwitchData 项目里真实落地的 AI 聊天流式接口，从控制器出口、服务层契约、HttpClient 流式调用、上游协议解析到落库时机，完整拆一遍这套实现。它不是「调个库」那么简单，中间几个坑很容易踩：迭代器里不能 catch 后 yield、SendAsync 默认会把响应体读完导致流式失效、上游明明是流式却可能回一个 JSON 错误包。

整体链路

整条链路是「浏览器 ← SSE ← 控制器 ← IAsyncEnumerable ← 上游流」：

flowchart LR
    A[浏览器 EventSource/fetch] -->|POST /api/AI/ChatStream| B[控制器 ChatStream]
    B -->|await foreach 逐行写| C[AiService.ChatStreamAsync]
    C -->|ResponseHeadersRead| D[上游 AI 平台]
    D -->|JSON 行流 RUNNING/SUCCESS| C
    C -->|data: {...}\n\n| B
    C -->|流结束后| E[PersistAsync 落库]

关键点只有一个：整条链路都必须「边产生边消费」，任何一环节做了全量缓冲，流式就退化成一次性返回。

一、控制器：SSE 的三个必备动作

控制器里没有 IActionResult，直接写 Response，因为流式响应一旦开始，就不能再返回统一的 JSON 包装了：

[Display(Name = "智能体管理（AI）-聊天问答（流式）")]
[HttpPost, Permission(PermissionPrivileges.AI.Chat)]
public async Task ChatStream([FromBody] ChatRequest request)
{
    Response.ContentType = "text/event-stream";
    Response.Headers.CacheControl = "no-cache";

    await foreach (var line in AiService.ChatStreamAsync(request, CurrentUserContext.UserId, RequestAborted))
    {
        await Response.WriteAsync(line, RequestAborted);
    }
}

三个细节值得说明：

ContentType = "text/event-stream"：告诉浏览器这是 SSE，fetch 也会按流读取；缺少它，浏览器会等整个响应结束才交付数据。
Cache-Control: no-cache：SSE 是长连接，中间若有代理或 CDN 缓存，事件会被攒起来批量下发，甚至直接不返回。
RequestAborted：ASP.NET Core 提供的一个 CancellationToken，客户端关闭页面、断网时它会被触发。把它一路传下去，才能及时停止对上游的读取，避免服务端白白占用一个连接和线程。

控制器里用的是 await foreach，这是消费 IAsyncEnumerable<T> 的标准写法，编译器会展开成 GetAsyncEnumerator + MoveNextAsync 的循环。

二、服务层：用 IAsyncEnumerable 定义「流」的契约

服务的返回类型是 IAsyncEnumerable<string>，而不是 Task<string>。这个选择很关键：Task 表达的是「未来某刻有一个值」，而流要表达的是「未来会有很多值，且值之间有先后依赖」。用 Task 就只能把数据攒进集合里再统一返回，流式的意义就没了。

public static async IAsyncEnumerable<string> ChatStreamAsync(
    ChatRequest request, string userId,
    [EnumeratorCancellation] CancellationToken cancellationToken = default)

[EnumeratorCancellation] 是这里的重点。异步迭代器的取消令牌有两条注入路径：一是方法参数直接传，二是调用方通过 WithCancellation(...) 传入。加上这个特性后，编译器生成的迭代器会把二者合并——如果调用方同时也传了 WithCancellation，以调用方的为准。没有它，await foreach 里的 WithCancellation 会被静默忽略，取消行为失灵。

迭代器的 catch 限制

写异步迭代器时最容易撞上的一条编译错误：yield return 不能出现在带 catch 的块里。C# 规范不允许在 catch 中产出值，因为异常处理期间迭代器状态机是不完整的。

但流式接口又必须把「连接失败」作为一条错误事件推给前端，不能直接抛异常（此时响应头可能已经发出去了）。解决办法是把异常先暂存成字符串，退出 try/catch 之后再 yield：

HttpResponseMessage upstream = null;
string connectError = null;
try
{
    upstream = await HttpHelper.PostJsonForStreamAsync(
        ApiAppSettings.AI.ChatBotUrl, JsonHelper.Serialize(upstreamRequest), headers, 30);
}
catch (Exception ex)
{
    Logger.Error($"[AI会话] 流式聊天接口连接失败：{ex.Message}", ex);
    connectError = ex.Message;
}

// 迭代器 catch 块内不能 yield，错误信息暂存后在此输出
if (connectError != null)
{
    yield return SseEvent(JsonHelper.Serialize(new { type = "error", message = $"AI接口调用失败: {connectError}" }));
    yield break;
}

这个模式可以推广到所有「异常需要转成事件」的场景：try 里只做、不产出；catch 里只记、不产出；产出统一放在 try/catch 之外。

yield break 则用来提前结束流。注意它和 return 的差别：yield break 只是告诉枚举器「没有更多元素了」，前面的 using、finally 依然会正常执行，资源不会泄漏。

三、HttpClient：让流式真正生效的开关

很多人写了 IAsyncEnumerable，前端却依然收不到增量数据，问题通常出在这一行：

// 流式响应：仅等待响应头到达即返回，避免整体读取响应体导致流式失效
var response = await httpClient.SendAsync(request, HttpCompletionOption.ResponseHeadersRead, cts.Token);
response.EnsureSuccessStatusCode();
return response;

SendAsync 默认的 HttpCompletionOption 是 ResponseContentRead，意思是「把响应体完整读进内存后再返回」。上游流式返回 30 秒的内容，这里就等 30 秒，然后一次性给你——流式彻底退化成同步调用，还白白多占一份内存。

改成 ResponseHeadersRead 后，只要响应头到达就返回 HttpResponseMessage，响应体还是「未读」状态，后续由调用方通过 ReadAsStreamAsync() 自己边读边处理。

还有一个容易混淆的点：这里的超时只覆盖「建立连接 + 拿到响应头」，30 秒。真正的业务耗时发生在读取响应体的阶段，不受这个超时约束——否则长回答一定会被误杀。这也是为什么这个辅助方法的参数名是 connectTimeoutSeconds 而不是 timeoutSeconds。

另外，超时产生的 OperationCanceledException 被包装成了 TimeoutException，这样上层能区分「我主动取消」和「连接超时」：

catch (OperationCanceledException ex) when (cts.Token.IsCancellationRequested)
{
    throw new TimeoutException("请求超时", ex);
}

注意 when (cts.Token.IsCancellationRequested) 这个异常过滤器——只有是自己这个 CTS 触发的取消才转成超时，调用方传入的取消不应该被误报成超时。

四、上游协议嗅探：流式请求也可能回 JSON

真实环境里，上游可能因为鉴权失败、参数校验不过、模型过载等原因，对一个 stream: true 的请求返回普通 JSON 错误包。如果不做判断，读取响应体的代码会把这段 JSON 当作 SSE 行来解析，最后得到「无内容」这种模糊错误。

所以拿到响应后先看 Content-Type：

var contentType = upstream.Content.Headers.ContentType?.MediaType ?? string.Empty;

// 上游返回JSON（错误或平台未按流式响应）：整包读取后按非流式逻辑处理
if (contentType.Contains("json", StringComparison.OrdinalIgnoreCase))
{
    var json = await upstream.Content.ReadAsStringAsync(cancellationToken);
    var upstreamResponse = JsonHelper.Deserialize<AiChatUpstreamResponse>(json);
    Logger.Info($"[AI会话] 流式请求返回非流式响应：{json}");

    if (upstreamResponse == null || upstreamResponse.Code != 0
        || upstreamResponse.FinishReason != "SUCCESS"
        || string.IsNullOrWhiteSpace(upstreamResponse.Data))
    {
        yield return SseEvent(JsonHelper.Serialize(new { type = "error", message = "AI平台返回异常" }));
        yield break;
    }

    yield return SseEvent(JsonHelper.Serialize(new { type = "meta", conversationId = upstreamResponse.ConversationId, messageId = upstreamResponse.MessageId }));
    yield return SseEvent(JsonHelper.Serialize(new { type = "delta", content = upstreamResponse.Data }));
    yield return SseEvent(JsonHelper.Serialize(new { type = "done", finishReason = upstreamResponse.FinishReason }));

    await PersistAsync(request, userId, upstreamResponse, (int)stopwatch.ElapsedMilliseconds);
    yield break;
}

这段的价值在于出口协议统一：不管上游是流式还是整包，前端收到的都是同一套 meta / delta / done / error 事件，前端不需要写两套分支。这是很典型的「在边界做协议适配」思路——把上游格式差异消化在服务层。

另外注意校验条件：Code != 0 || FinishReason != "SUCCESS" || Data 为空，三者任一不满足即失败。只看 Code 是不够的，上游有些错误是以 finish_reason 表达的；而 Data 为空也可能是一种「静默失败」。宁可严格一点，也不要让前端拿到一个空回答还以为成功了。

五、逐行解析：兼容两种格式，过滤中间过程

真正读取流的部分，用的是 StreamReader.ReadLineAsync(cancellationToken)：

await using var stream = await upstream.Content.ReadAsStreamAsync(cancellationToken);
using var reader = new StreamReader(stream, Encoding.UTF8);

while (true)
{
    var line = await reader.ReadLineAsync(cancellationToken);
    if (line == null) break;

    var payload = line.Trim();
    if (payload.Length == 0) continue;

    // 兼容SSE前缀（data: {...}）与纯JSON行两种格式
    if (payload.StartsWith("data:", StringComparison.Ordinal)) payload = payload[5..].Trim();
    if (!payload.StartsWith("{", StringComparison.Ordinal)) continue;

    AiChatUpstreamResponse chunk;
    try
    {
        chunk = JsonHelper.Deserialize<AiChatUpstreamResponse>(payload);
    }
    catch (Exception)
    {
        continue; // 非法JSON行忽略
    }
    if (chunk == null) continue;
    // ...
}

这里有四个容错点：

空行跳过。SSE 里空行是事件分隔符，不是数据。
data: 前缀可选。标准 SSE 一定有 data: 前缀，但很多自研网关直接按 NDJSON（每行一个 JSON）返回。两种都要能认，用 [5..] 这个范围切片比 Substring(5) 更简洁。
非 { 开头直接忽略。这样既过滤了 event:、id:、注释行（: ping），也不会浪费一次反序列化。
单行解析失败不中断整条流。一条坏行不应该让整个回答作废，catch 后 continue 即可。

阶段推进式协议：RUNNING 与 SUCCESS

这个上游不是 OpenAI 那种「每个 token 一个 delta」的格式，而是阶段推进式：每行是一个完整 JSON，finish_reason 为 RUNNING 时 data 里装的是中间过程（问题回显、路由标记、知识检索结果，单次可达几十 KB），只有 SUCCESS 那一行才是最终完整回复。

于是解析逻辑变成「边收集元信息、边丢弃中间过程」：

// 首个含会话ID的行即下发元信息，后续行补齐消息ID
if (meta == null && !string.IsNullOrEmpty(chunk.ConversationId))
{
    meta = chunk;
    yield return SseEvent(JsonHelper.Serialize(new { type = "meta", conversationId = meta.ConversationId, messageId = meta.MessageId }));
}
else if (meta != null && string.IsNullOrEmpty(meta.MessageId) && !string.IsNullOrEmpty(chunk.MessageId))
{
    meta.MessageId = chunk.MessageId;
}

// RUNNING行为中间过程，不下发前端；SUCCESS行为最终完整回复
if (chunk.FinishReason == "SUCCESS")
{
    meta ??= chunk;
    meta.Data = chunk.Data;
    meta.FinishReason = chunk.FinishReason;
    break;
}

两个设计取舍：

元信息先发。前端拿到 conversationId 就可以立刻建会话、渲染输入框，不必等回答结束。messageId 可能在后续行才出现，所以用 else if 做补齐，而不是重新发一条 meta——避免前端重复处理。
中间过程不下发。这些内容对用户是噪音，而且体积大，推给前端只会让「流式」看起来像卡顿。如果将来要做「思考过程展示」，这里的过滤条件就是唯一的开关点。

break 之后循环结束，后面的 stopwatch.Stop()、内容校验、delta / done 事件、落库都会执行——因为 break 跳出的是 while，不是 using 作用域。

六、SSE 事件的拼装

出口统一用一个私有方法生成事件行：

/// <summary>构造SSE事件行（data: 前缀 + 空行作为事件边界）</summary>
private static string SseEvent(string json) => $"data: {json}\n\n";

末尾的 \n\n 不能省：SSE 规范中，两个换行才构成一个事件的结束。只写一个 \n 的话，浏览器会把后续事件当作同一个事件的延续，前端只会收到一串拼在一起的 JSON。

事件类型保持在四个：

type	时机	载荷
`meta`	拿到 conversationId 时	conversationId、messageId
`delta`	拿到最终回复时	content
`done`	流正常结束	finishReason
`error`	连接失败/上游异常/无内容	message

前端只处理这四种，就不用关心上游到底是流式、整包，还是中途报了错误码。

七、资源释放与落库时机

using (upstream) 和 await using var stream 两层都要有：

HttpResponseMessage 不释放，底层连接不会归还连接池，长时间运行必然耗尽。
NetworkStream 不释放，服务端会一直认为这个连接还在读。

而且 using 的作用域是一个兜底：即使调用方 await foreach 在中途 break（比如前端断连、异常），迭代器被 Dispose 时 using 依然会执行，ReadAsStreamAsync 拿到的流会被关闭，上游连接随之释放。这是 IAsyncEnumerable 相比手写回调的隐性好处——取消与清理是语言级保证的，不靠开发者记性。

落库放在流的最后：

yield return SseEvent(JsonHelper.Serialize(new { type = "delta", content = answer }));
yield return SseEvent(JsonHelper.Serialize(new { type = "done", finishReason = meta.FinishReason ?? "SUCCESS" }));

if (string.IsNullOrEmpty(meta.ConversationId))
{
    Logger.Error("[AI会话] 流式响应缺少会话ID，跳过落库");
    yield break;
}

await PersistAsync(request, userId, meta, (int)stopwatch.ElapsedMilliseconds);

顺序很重要：先推 done，再落库。落库要写会话表、消息表，可能几百毫秒；如果放在 done 之前，前端会多等一截，流式的体验就打了折扣。而 PersistAsync 内部整体 try/catch，把异常全部吞掉只记日志——落库失败不能影响已经推送成功的回答，否则用户看到完整回答却又收到一个错误。

catch (Exception ex)
{
    Logger.Error($"[AI会话] 会话及消息落库失败：conversationId={upstream.ConversationId}", ex);
}

八、踩坑清单

把上面这些浓缩成一份可复用的检查表：

ContentType 没设成 text/event-stream，浏览器全程静默。
漏了 Cache-Control: no-cache，被代理缓冲。
SendAsync 用了默认的 ResponseContentRead，流式直接失效。
连接超时和读取超时混成一个，长回答被误杀。
迭代器里 catch 后直接 yield return，编译不过。
没有 [EnumeratorCancellation]，WithCancellation 静默失效。
一个 \n 当事件结束符，前端收到粘在一起的 JSON。
只认 data: 前缀，遇到 NDJSON 上游全部丢弃。
只校验 code == 0，漏掉 finish_reason 与空 data 的静默失败。
落库放在 done 之前，白等几百毫秒。
不判断上游 Content-Type，把 JSON 错误包当流解析。

总结

这套实现的核心可以概括成三句话：

契约层用 IAsyncEnumerable<string> 表达「多值 + 顺序 + 可取消」，而不是用 Task 假装流式；
传输层必须 ResponseHeadersRead + 逐行读取，任何一次全量缓冲都会让流式名存实亡；
边界层做协议适配（JSON 包降级、data: 前缀兼容、阶段性协议过滤），让前端只面对固定的 meta/delta/done/error 四种事件。

把这三点想清楚，SSE 就不只是「换个 Content-Type」，而是一条从上游到浏览器、能被取消、能被兜底、失败不污染主链路的完整通道。