基于Flask+Gunicorn+Gevent的在线用户列表实现咨询

核心思路 #

既然你已经用了flask-socketio，在线用户管理的核心是把Socket连接会话（sid）和已登录用户绑定，结合共享存储（避免多worker内存隔离问题）维护在线状态，再通过Socket事件实时推送列表。

步骤1：配置依赖与共享存储 #

首先确保安装所需依赖：

pip install flask-socketio redis

因为你用Gunicorn多worker模式，普通内存变量无法跨worker共享，所以用Redis同时做SocketIO消息队列和在线用户存储。

步骤2：关联Flask-Login与SocketIO连接 #

在SocketIO的connect/disconnect事件中，通过Flask-Login的current_user获取登录用户，将用户ID与Socket会话ID（sid）绑定到Redis：

from flask import Flask
from flask_login import LoginManager, UserMixin, current_user, login_user
from flask_socketio import SocketIO, emit, disconnect
import redis

app = Flask(__name__)
app.secret_key = 'your-secure-secret-key'

# 配置Flask-Login
login_manager = LoginManager(app)
login_manager.login_view = 'login'

# 配置Redis（SocketIO消息队列+在线用户存储）
redis_url = 'redis://localhost:6379/0'
socketio = SocketIO(app, message_queue=redis_url)
redis_client = redis.Redis.from_url(redis_url)

# 模拟用户模型（替换成你实际的用户查询逻辑）
class User(UserMixin):
    def __init__(self, user_id, username):
        self.id = user_id
        self.username = username

users_db = {1: User(1, "Alice"), 2: User(2, "Bob")}

@login_manager.user_loader
def load_user(user_id):
    return users_db.get(int(user_id))

# 示例登录路由（你已有自己的登录逻辑可忽略）
@app.route('/login/<int:user_id>')
def login(user_id):
    user = users_db.get(user_id)
    if user:
        login_user(user)
        return f"Logged in as {user.username}"
    return "User not found"

# Socket连接事件：验证用户并记录在线状态
@socketio.on('connect')
def handle_connect():
    if not current_user.is_authenticated:
        disconnect()
        return
    # 存储用户ID与Socket会话的映射
    redis_client.hset('online_users', current_user.id, request.sid)
    # 推送更新后的在线列表
    update_online_users()

# Socket断开事件：移除用户在线记录
@socketio.on('disconnect')
def handle_disconnect():
    if current_user.is_authenticated:
        redis_client.hdel('online_users', current_user.id)
        update_online_users()

# 通用函数：获取在线用户并广播
def update_online_users():
    online_user_ids = redis_client.hkeys('online_users')
    # 替换成你实际的用户信息查询逻辑
    online_users = [users_db[int(uid)].username for uid in online_user_ids]
    # 广播给所有连接的客户端
    emit('update_online_users', online_users, broadcast=True)

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

步骤3：客户端监听实时更新 #

在前端页面添加SocketIO监听，接收服务器推送的在线列表并更新DOM：

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
<ul id="online-users"></ul>
<script>
    const socket = io();
    socket.on('update_online_users', (users) => {
        const userList = document.getElementById('online-users');
        userList.innerHTML = '';
        users.forEach(username => {
            const li = document.createElement('li');
            li.textContent = username;
            userList.appendChild(li);
        });
    });
</script>

关键注意事项 #

• Gunicorn多worker适配：必须用Redis这类共享存储，不能用Python全局字典——每个Gunicorn worker有独立内存空间，全局变量无法跨worker同步。
• Flask-Session配置：确保你的flask-session用服务器端存储（比如Redis），否则多worker下用户会话无法共享，current_user会失效。
• 启动命令：保持你原来的启动命令即可，确保Redis服务处于运行状态：

gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker --bind 127.0.0.1:$port app:app

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