← 返回首页

SSE 在 Flutter 中的应用与踩坑记录

SSE 在 Flutter 中的应用与踩坑记录

1. SSE 是什么

SSE（Server-Sent Events）是一种基于 HTTP 的服务器推送协议。客户端发起一次长连接请求，服务端以文本流的方式持续返回事件；客户端不需要反复发起请求，就可以接收任务进度、通知或增量结果。

SSE 使用 text/event-stream 作为响应类型。一个事件由若干行字段组成，以空行结束：

id: 42
event: progress
data: {"status":"running","percent":60}

常见字段包括：

字段 作用
event 事件名称，未指定时可视为默认消息类型
data 事件数据，可以出现多次，客户端按换行合并
id 事件编号，可用于断线恢复
retry 服务端建议的重连等待时间（毫秒）

以冒号开头的行是注释，通常用于心跳保活。事件之间必须使用空行分隔；只有读到完整事件后，客户端才能交给业务层解析。

SSE 与其他方案的区别

方案 通信方向 连接模型 适合场景
普通 HTTP 请求-响应 短连接 查询、提交表单、一次性下载
轮询 客户端主动查询 多次短连接 低频状态变化、实现成本敏感的场景
SSE 服务端单向推送 一个长连接 任务进度、实时日志、通知、流式文本
WebSocket 双向通信 一个长连接 聊天、协同编辑、实时控制

SSE 的优势是基于现有 HTTP 基础设施，浏览器和代理支持较好，服务端实现简单；限制是通信方向主要是服务端到客户端，且移动网络切换、后台挂起和代理超时需要额外处理。

sse-flow-diagram

2. 什么场景适合使用 SSE

2.1 长任务的阶段性进度

例如文件处理、报表生成、媒体转码、AI 推理等任务。客户端提交任务后，服务端可以连续发送 queued、running、success 或 failed 等状态，页面据此更新进度和提示。

这类场景的关键是：服务端事件应当表达「状态变化」，而不只是定时重复发送相同数据。客户端还需要定义明确的终态，收到终态后立即关闭连接。

2.2 流式文本或日志

服务端可以不断发送新增文本、日志行或结构化片段，让用户尽早看到结果。接入前必须约定数据语义：

  • delta（增量）：每条消息只包含新增内容，客户端可以追加。
  • full snapshot（全量快照）：每条消息包含当前完整内容，客户端必须做快照合并，不能盲目拼接。

这两个语义看起来相似，却会直接决定客户端是否出现重复、倒退或丢字。

2.3 低频实时通知

通知、任务完成提醒、后台状态变化等场景通常只要求「有变化就推送」，不需要客户端和服务端双向对话，SSE 比 WebSocket 更轻量。

2.4 不适合使用 SSE 的情况

  • 需要客户端持续向服务端发送高频数据，例如实时音视频控制。
  • 需要严格的双向低延迟交互，例如多人协同光标。
  • 事件量极大且不允许文本编码开销，或需要自定义二进制协议。
  • 业务无法接受断线重连后的重复消息，也没有设计幂等或恢复游标。

3. Flutter 中的基本实现

Flutter 没有内置统一的 SSE 客户端，通常使用 Dio、http 等网络库取得字节流，再自行解析协议。下面示例使用 Dio：

import 'dart:convert';
import 'package:dio/dio.dart';

Stream<Map<String, dynamic>> openSse(Dio dio, String url) async* {
  final response = await dio.get<ResponseBody>(
    url,
    options: Options(
      responseType: ResponseType.stream,
      headers: {
        'Accept': 'text/event-stream',
        'Cache-Control': 'no-cache',
      },
      // 长连接不设置整体接收超时，空闲超时另行处理。
      receiveTimeout: Duration.zero,
    ),
  );

  final lines = response.data!.stream
      .transform(utf8.decoder)
      .transform(const LineSplitter());
  final dataLines = <String>[];

  await for (final line in lines) {
    if (line.isEmpty) {
      if (dataLines.isNotEmpty) {
        final data = dataLines.join('\n');
        dataLines.clear();
        if (data.trim() != '[DONE]') {
          final value = jsonDecode(data);
          if (value is Map<String, dynamic>) yield value;
        }
      }
      continue;
    }

    if (line.startsWith('data:')) {
      var value = line.substring(5);
      if (value.startsWith(' ')) value = value.substring(1);
      dataLines.add(value);
    }
  }
}

这个示例只展示最小路径。生产代码建议拆成三层：

  1. Transport 层：负责 HTTP 请求、响应状态码、请求头和取消令牌。
  2. Parser 层：负责 chunk、UTF-8、换行、空 frame、多行 data 和注释心跳。
  3. Session 层：负责生命周期、重连、终态、错误分类和业务模型转换。

业务侧订阅生命周期流，而不是只订阅数据流，会更容易处理连接中、重试中和已关闭状态：

final session = SseSession<TaskEvent>(
  request: request,
  decode: TaskEvent.fromJson,
);

final subscription = session.events.listen((event) {
  switch (event.type) {
    case SseEventType.data:
      render(event.data!);
    case SseEventType.retrying:
      showRecovering(event.attempt);
    case SseEventType.error:
      showError(event.error);
    case SseEventType.closed:
      hideLoading();
    default:
      break;
  }
});

// 页面销毁、任务取消或用户离开时调用。
await session.close();
await subscription.cancel();

4. 生产化设计要点

4.1 明确终态，不要把断流当成功

网络流正常结束，只能说明 TCP/HTTP 流关闭，不能证明业务任务完成。建议同时定义两类结束信号：

  • 协议结束：例如 [DONE]。
  • 业务终态：例如 success、failed、canceled。

只有收到明确终态才停止重连并关闭会话；未收到终态就断流，应进入恢复或错误流程。

4.2 重连使用指数退避与抖动

可以使用 1s、2s、4s、8s 的指数退避，并设置最大等待时间和最大次数。加入少量随机抖动，避免服务端故障恢复时大量客户端同时重连。

重连前要先取消旧连接。每次连接分配一个 generation（代际编号），回调只处理当前代际的数据，防止旧连接的异步回调串入新连接。

4.3 断线恢复必须幂等

如果协议支持 id，重连时可携带 Last-Event-ID；如果业务使用任务 ID，则服务端应返回当前完整状态，客户端按任务 ID 重新订阅。无论采用哪种方式，都要允许重复事件到达：

  • 用事件 ID 去重，或
  • 让状态更新具备幂等性，或
  • 使用版本号/序列号丢弃旧快照。

4.4 空闲超时与心跳

长连接不应设置一个很短的整体接收超时，否则任务越久越容易被误判失败；但完全没有空闲超时，又无法识别代理或网络已经「假连接」。更稳妥的做法是：整体接收时长不限制，只限制两个数据 chunk 之间的空闲时间，并让服务端定期发送注释心跳。

4.5 错误要分层

建议至少区分：

  • HTTP 错误：408、429、5xx 通常可以重试；认证失败、参数错误等 4xx 通常不应重试。
  • 网络错误：断网、切网、连接重置，可按策略恢复。
  • 解析错误：JSON 非法、字段类型不匹配，通常说明协议不兼容，不应盲目重连。
  • 业务错误：服务端返回失败状态，应展示业务提示并结束任务。

5. 踩坑记录

5.1 把 chunk 当成完整事件

底层网络流的一个 chunk 可能只包含半行、半个 UTF-8 字符，或者同时包含多个事件。不能对每个 chunk 直接 jsonDecode。正确顺序是「字节流 → UTF-8 解码 → 按行拆分 → 按空行组 frame → 合并 data → JSON 解析」。

5.2 忽略多行 data

SSE 允许一个事件出现多行 data:。如果只读取第一行，长文本或格式化 JSON 会被截断。解析时应收集所有 data 行，并用换行符合并。

5.3 用 taskStatus=success 过早关闭

有些服务会先发送「某个阶段完成」，之后再发送整个任务的 complete 事件。如果通用层只看一个状态字段就关闭连接，后续终态可能永远到不了客户端。业务模型应覆盖通用的 isTerminal 判断，按真正的业务终态结束。

5.4 把 full snapshot 当 delta

这是流式 UI 最常见的问题之一。收到更长快照时只播放新增后缀；收到更短且属于旧内容的快照时应忽略；无前缀关系时按一次整体改写处理。不要默认执行 old + incoming。

5.5 只处理 onDone，不处理异常关闭

onDone 可能发生在服务端主动结束、代理断开、网络切换或客户端取消之后。必须结合「是否收到终态」「是否主动关闭」「当前连接代际」判断下一步，不能把所有 onDone 都当成功。

5.6 忘记释放连接和订阅

页面退出时如果只取消 UI 的 StreamSubscription，底层 HTTP 长连接仍可能存在。应提供幂等的 close()，同时取消 CancelToken、frame 订阅、重连计时器和事件控制器。Flutter 页面销毁、应用切后台、用户取消任务都应触发相应处理。

5.7 代理或网关缓冲导致「看起来不是流式」

即使服务端逐条发送，反向代理也可能积累到一定大小后才转发。需要检查响应头、禁用不必要的缓冲、发送心跳，并在真实网络环境验证首字节时间和事件间隔。仅在本地直连环境测试不足以证明流式体验。

5.8 日志泄露敏感信息

SSE 请求通常包含 Token、任务标识、用户输入或生成内容。日志应只记录连接代际、事件类型、序列号、耗时和错误分类；对 URL 查询参数、请求体、原始 data 做脱敏或截断，禁止把完整鉴权头写入日志。

6. 测试与排查清单

上线前至少覆盖以下用例：

  • 一行一个事件、多个事件粘连、事件跨 chunk、最后一帧无空行。
  • 多行 data、注释心跳、空事件、未知字段、[DONE]。
  • 首次连接失败、流中途断开、空闲超时、重连次数耗尽。
  • 重连后重复事件、乱序事件、旧连接回调晚到。
  • 业务成功、失败、取消以及未收到终态就断流。
  • 页面退出、应用切后台、网络切换后的取消与恢复。
  • 代理环境下的首字节时间、事件延迟和心跳是否可达。

排查问题时，建议记录一条不含敏感内容的链路日志：连接开始 → HTTP 状态 → 首个事件 → 最近事件时间 → 重连次数 → 终态/关闭原因。这样既能区分服务端没有发送、网络没有转发、客户端没有解析，还是 UI 没有消费，也不会把业务数据写进日志。

7. 总结

SSE 的核心价值是用一个普通 HTTP 长连接，把「服务端状态变化」及时送到客户端。它适合单向推送、阶段性进度和流式文本，不适合高频双向交互。

在 Flutter 中，真正困难的部分不是把响应声明为 ResponseType.stream，而是把协议解析、连接生命周期、重连恢复、业务终态和 UI 展示边界设计清楚。实践中可以遵循以下原则：

  1. 先确认服务端返回的是 delta 还是 full snapshot。
  2. 按协议组装完整 frame，再解析 JSON。
  3. 明确业务终态，未收到终态的断流不能当成功。
  4. 重连要有退避、上限、幂等和旧连接隔离。
  5. 空闲超时与心跳配合使用，兼顾假连接检测和长任务时长。
  6. 页面销毁时彻底关闭 session，避免连接、定时器和订阅泄漏。
  7. 用真实网络、代理和移动端生命周期验证，静态代码检查不能替代运行时证据。

当这些边界被落实后，SSE 就不只是「能收到消息」的技巧，而会成为一套可观测、可恢复、可维护的实时任务通信方案。

标签：