如何用Dio结合SSE实时数据构建并更新Flutter ListView

核心问题分析 #

1. SSE数据流未正确传递到UI层：原ChatApi仅监听SSE但未暴露数据流，导致UI只能获取一次数据，无法持续接收更新。
2. 重复创建SSE连接：每次build调用getChatsList都会新建Stream，触发重复请求。
3. 未处理SSE连接断开：页面退出时未主动取消订阅，导致连接残留。

步骤1：修正ChatApi，返回SSE数据流 #

将原Future返回改为Future<Stream<dynamic>>，暴露SSE的持续数据流：

class ChatApi{
  late Dio dioRest;

  ChatApi() {
    dioRest = Dio(
      BaseOptions(
        headers: {
          HttpHeaders.contentTypeHeader: ContentType.json.value,
        },
      ),
    );
  }

  Future<Stream<dynamic>> getRoomChatList(int roomId, int categoryId) async {
    try {
      final token = getToken();
      final Response<ResponseBody> result = await dioRest.post(
        '${token.baseUrl}..../getRoomChatList',
        options: Options(...), // 保留原配置
        queryParameters: {'roomId': roomId, 'categoryId': categoryId},
      );

      if (result.data?.stream == null) {
        throw DioError(requestOptions: result.requestOptions);
      }

      StreamTransformer<Uint8List, List<int>> unit8Transformer = StreamTransformer.fromHandlers(
        handleData: (data, sink) {
          sink.add(List<int>.from(data));
        },
      );

      // 转换SSE事件为JSON数据，返回数据流
      return result.data!.stream
          .transform(unit8Transformer)
          .transform(const Utf8Decoder())
          .transform(const LineSplitter())
          .transform(const SseTransformer())
          .map((event) {
            print("Event: ${event.id}, ::  ${event.data}");
            return jsonDecode(event.data);
          });
    } on DioError catch (error) {
      if (error.response?.statusCode != null && error.response!.statusCode! >= 400) {
        throw TokenException(message: "Token invalid or expired");
      }
      throw BusinessException(BusinessError.fromJson(error.response?.data), statusCode: error.response?.statusCode);
    } catch (error) {
      rethrow;
    }
  }
}

步骤2：修正ChatController，转发完整数据流 #

使用async*和yield*转发SSE数据流，确保UI能持续接收更新：

class ChatController {
  ChatController({
    required ChatApi chatApi,
    required ProviderRef ref,
  })  : _chatApi = chatApi,
        _ref = ref;

  final ChatApi _chatApi;
  final ProviderRef _ref;

  Stream<dynamic> getChatsList(int roomId, int categoryId) async* {
    final stream = await _chatApi.getRoomChatList(roomId, categoryId);
    yield* stream; // 转发完整数据流
  }
}

步骤3：添加Riverpod StreamProvider管理数据流 #

用StreamProvider.family根据房间ID和分类ID缓存数据流，避免重复请求，并自动管理订阅生命周期：

final chatListStreamProvider = StreamProvider.family<Map<String, dynamic>, (int roomId, int categoryId)>((ref, args) async* {
  final chatController = ref.watch(chatControllerProvider);
  final stream = await chatController.getChatsList(args.$1, args.$2);
  yield* stream;
});

步骤4：修正ChatScreen，优化UI逻辑与连接管理 #

使用Riverpod的when方法处理数据流状态，页面退出时自动取消SSE订阅：

class ChatScreen extends ConsumerStatefulWidget {
  final dynamic currentUser;
  final dynamic data;
  final dynamic screen;
  final dynamic selCategory;

  const ChatScreen({
    Key? key,
    required this.currentUser,
    required this.data,
    required this.screen,
    required this.selCategory,
  }) : super(key: key);

  @override
  ConsumerState<ChatScreen> createState() => _ChatScreenState();
}

class _ChatScreenState extends ConsumerState<ChatScreen> {
  dynamic dataItem;
  List<dynamic>? chatList;

  @override
  Widget build(BuildContext context) {
    final roomId = widget.data['id'] as int;
    final categoryId = widget.selCategory['id'] as int;

    // 监听数据流，Riverpod自动管理订阅
    final chatStream = ref.watch(chatListStreamProvider((roomId, categoryId)));

    return WillPopScope(
      onWillPop: () async {
        return true;
      },
      child: chatStream.when(
        loading: () => const Loader(),
        error: (error, stackTrace) {
          return Center(child: Text('加载失败: $error'));
        },
        data: (data) {
          dataItem = data['chatRoom'];
          chatList = data['chatList'];

          if (chatList?.isEmpty ?? true) {
            return const NoChat();
          }

          return showChatView();
        },
      ),
    );
  }

  Widget showChatView() {
    return ListView.builder(
      reverse: true, // 聊天列表默认倒序显示最新消息
      itemCount: chatList!.length,
      itemBuilder: (context, index) {
        final chat = chatList![index];
        // 替换为你的自定义聊天项Widget
        return ListTile(
          title: Text(chat['content'] ?? ''),
          subtitle: Text(chat['sender'] ?? ''),
        );
      },
    );
  }
}

关键优化点 #

1. 数据流持续传递：从ChatApi到ChatController再到UI层，完整保留SSE的实时数据流。
2. 自动管理连接生命周期：Riverpod的StreamProvider会在页面销毁时自动取消SSE订阅，断开连接。
3. 避免重复请求：通过family参数缓存数据流，同一房间/分类不会重复发起SSE连接。
4. 清晰的状态处理：用when方法统一处理加载、错误、数据三种状态，替代原StreamBuilder的冗余逻辑。

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