如何在Flutter中流式处理支持Server-Sent Events(SSE)的API响应？

一、先确认 Python 服务器的 SSE 格式合规性 #

SSE 有固定协议格式，每个数据块必须以data: 开头，以\n\n结尾。你的 Python 服务器需要按这个格式返回内容，示例代码如下：

from flask import Flask, Response
import time

app = Flask(__name__)

@app.route('/stream')
def stream_response():
    def generate():
        # 模拟逐段生成响应内容
        content_blocks = ["你好", "这是流式响应示例", "像 ChatGPT 一样", "逐步显示内容"]
        for block in content_blocks:
            yield f'data: {block}\n\n'
            time.sleep(1)  # 模拟处理延迟
    return Response(generate(), mimetype='text/event-stream')

if __name__ == '__main__':
    app.run(debug=True)

二、Flutter 端实现方案 #

方案1：用原生 HttpClient 手动解析 SSE #

无需第三方库，直接用 Flutter 自带的网络工具实现：

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'package:flutter/material.dart';

class SSEStreamDemo extends StatefulWidget {
  const SSEStreamDemo({super.key});

  @override
  State<SSEStreamDemo> createState() => _SSEStreamDemoState();
}

class _SSEStreamDemoState extends State<SSEStreamDemo> {
  String _displayContent = '';
  StreamSubscription? _streamSub;
  late HttpClient _httpClient;

  @override
  void initState() {
    super.initState();
    _httpClient = HttpClient();
    _startStream();
  }

  void _startStream() async {
    try {
      final request = await _httpClient.getUrl(Uri.parse('http://你的服务器地址:5000/stream'));
      // 设置 SSE 专属请求头
      request.headers.add('Accept', 'text/event-stream');
      request.headers.add('Cache-Control', 'no-cache');
      
      final response = await request.close();
      // 监听响应流，逐行解析
      _streamSub = response
          .transform(utf8.decoder)
          .transform(const LineSplitter())
          .listen((line) {
        if (line.startsWith('data: ')) {
          final content = line.substring(6).trim();
          if (content.isNotEmpty) {
            setState(() => _displayContent += content);
          }
        }
      }, onError: (err) => print('请求出错: $err'), onDone: () => print('流结束'));
    } catch (e) {
      print('连接失败: $e');
    }
  }

  @override
  void dispose() {
    _streamSub?.cancel();
    _httpClient.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('SSE 流式展示')),
      body: Padding(
        padding: const EdgeInsets.all(16.0),
        child: Text(_displayContent, style: const TextStyle(fontSize: 16)),
      ),
    );
  }
}

方案2：用第三方库 sse_client 简化开发 #

如果不想手动解析协议，可使用现成库：

1. 在pubspec.yaml添加依赖：

dependencies:
  sse_client: ^2.0.0

2. 实现代码：

import 'package:flutter/material.dart';
import 'package:sse_client/sse_client.dart';

class SSELibraryDemo extends StatefulWidget {
  const SSELibraryDemo({super.key});

  @override
  State<SSELibraryDemo> createState() => _SSELibraryDemoState();
}

class _SSELibraryDemoState extends State<SSELibraryDemo> {
  String _displayContent = '';
  SseClient? _sseClient;

  @override
  void initState() {
    super.initState();
    _connectSSE();
  }

  void _connectSSE() {
    _sseClient = SseClient.connect(
      Uri.parse('http://你的服务器地址:5000/stream'),
      headers: {
        'Accept': 'text/event-stream',
        'Cache-Control': 'no-cache',
      },
    );

    _sseClient?.stream.listen((event) {
      if (event.data != null && event.data!.isNotEmpty) {
        setState(() => _displayContent += event.data!);
      }
    }, onError: (err) => print('错误: $err'), onDone: () => print('流结束'));
  }

  @override
  void dispose() {
    _sseClient?.close();
    super.dispose();
  }

  @override
  Widget build(BuildContext context) {
    return Scaffold(
      appBar: AppBar(title: const Text('SSE 库实现')),
      body: Padding(
        padding: const EdgeInsets.all(16.0),
        child: Text(_displayContent, style: const TextStyle(fontSize: 16)),
      ),
    );
  }
}

三、关键注意事项 #

• 移动端网络权限：Android 在AndroidManifest.xml添加<uses-permission android:name="android.permission.INTERNET"/>；iOS 若用 HTTP 非加密连接，需在Info.plist配置NSAppTransportSecurity允许非 HTTPS 请求。
• Flutter Web 并非无法实现，之前失败大概率是跨域问题，需在 Python 服务器配置 CORS，比如给 Flask 添加flask-cors库并启用跨域。
• 若要更优雅的状态管理，可将流转换为Stream<String>，用StreamBuilder替代setState更新 UI。

内容的提问来源于stack exchange，提问作者MeulenCV