如何在Flutter应用与Flask API间通过流监听用户状态变化？

一、方案可行性说明 #

你的方案完全可行，Flask可通过**Server-Sent Events (SSE)**实现用户状态的持续推送，Flutter端配合StreamBuilder监听SSE流，就能动态根据用户状态切换页面。

二、Flask端实现（SSE方式） #

1. 依赖准备 #

需要用到flask和redis（跨路由共享状态变更，解决Flask多线程环境下的消息传递问题），先安装依赖：

pip install flask redis

2. 核心代码实现 #

from flask import Flask, Response, jsonify
import redis
import json
from time import sleep

app = Flask(__name__)
# 初始化Redis连接，用于状态存储和消息发布订阅
r = redis.Redis(host='localhost', port=6379, db=0)

@app.route('/stream')
def stream_user_state():
    def generate_sse():
        # 订阅用户状态更新频道
        pubsub = r.pubsub()
        pubsub.subscribe('user_state_updates')
        
        # 先推送当前已存储的用户状态（初始化用）
        current_user = r.get('current_user')
        if current_user:
            yield f'data: {current_user.decode()}\n\n'
        
        # 持续监听频道内的状态变更消息
        for message in pubsub.listen():
            if message['type'] == 'message':
                yield f'data: {message["data"].decode()}\n\n'
    
    # 设置SSE专属响应头，保持长连接
    return Response(generate_sse(), mimetype='text/event-stream')

@app.route('/login', methods=['POST'])
def login():
    # 模拟登录逻辑，生成用户状态数据
    current_user = {
        'id': 1,
        'username': 'test_user',
        'is_verified': True,
        'is_logged_in': True
    }
    # 存储状态到Redis，并发布变更消息到频道
    r.set('current_user', json.dumps(current_user))
    r.publish('user_state_updates', json.dumps(current_user))
    return jsonify({'status': 'success'})

@app.route('/logout')
def logout():
    # 模拟登出逻辑，清除状态并发布null值
    r.delete('current_user')
    r.publish('user_state_updates', json.dumps(None))
    return jsonify({'status': 'success'})

3. 跨路由状态推送逻辑 #

通过Redis的发布/订阅机制，任何修改用户状态的路由（比如/login、/logout），只需向user_state_updates频道发布新状态，/stream路由就能实时捕获并推送给客户端。

三、Flutter端实现 #

1. 依赖配置 #

在pubspec.yaml中添加http包：

dependencies:
  flutter:
    sdk: flutter
  http: ^1.1.0

2. 封装用户状态流服务 #

import 'dart:async';
import 'dart:convert';
import 'package:http/http.dart' as http;

class UserStateStream {
  static Stream<CurrentUser?> get stream {
    // 用周期性任务模拟SSE监听，移动端SSE长连接易中断，轮询更稳定
    return Stream.periodic(const Duration(seconds: 1), (_) async {
      final response = await http.get(Uri.parse('http://你的Flask服务器地址/stream'));
      if (response.statusCode == 200) {
        // 解析SSE格式的数据（每行以data:开头）
        final lines = response.body.split('\n');
        for (var line in lines) {
          if (line.startsWith('data:')) {
            final data = line.substring(5).trim();
            if (data == 'null') {
              return null;
            }
            return CurrentUser.fromJson(jsonDecode(data));
          }
        }
      }
      return null;
    }).asyncMap((event) => event);
  }
}

// 用户状态模型类
class CurrentUser {
  final int id;
  final String username;
  final bool isVerified;
  final bool isLoggedIn;

  CurrentUser({
    required this.id,
    required this.username,
    required this.isVerified,
    required this.isLoggedIn,
  });

  factory CurrentUser.fromJson(Map<String, dynamic> json) {
    return CurrentUser(
      id: json['id'],
      username: json['username'],
      isVerified: json['is_verified'],
      isLoggedIn: json['is_logged_in'],
    );
  }
}

3. StreamBuilder集成代码 #

@override
Widget build(BuildContext context) {
  return Scaffold(
    body: StreamBuilder<CurrentUser?>(
      stream: UserStateStream.stream,
      builder: (context, snapshot) {
        if (snapshot.connectionState == ConnectionState.waiting) {
          return const Center(child: CircularProgressIndicator());
        }
        if (snapshot.hasError) {
          print(snapshot.error);
          return const Center(child: Text('加载出错'));
        }
        if (snapshot.hasData && snapshot.data!.isLoggedIn) {
          // 根据用户状态跳转对应页面，示例为主页
          return const HomePage();
        } else {
          return const LoginPage();
        }
      },
    ),
  );
}

四、替代方案说明 #

如果移动端SSE长连接不稳定，可改用以下方案：

• WebSocket双向通信：Flask端用flask-socketio，Flutter端用socket_io_client包，实现更可靠的双向状态同步
• 调整轮询间隔：根据业务需求缩短或延长轮询时间，平衡实时性和资源消耗

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