在使用 Flask 开发的 Web 应用系统中，要获取**实时连接到系统的客户端信息**（如：当前会话数量、每个客户端的 IP、访问页面、会话持续时间等），需要注意以下几点： - HTTP 是无状态协议，Flask 本身不直接提供“实时连接”监控。 - 所谓“客户端连接”，通常是指“活跃会话”或“近期有请求的用户”。 - 我们可以通过 **会话管理 + 请求钩子 + 内存/数据库记录** 来实现近似“实时”的客户端行为追踪。 --- ### ✅ 解决方案概述 我们将使用以下技术组合来实现目标： 1. 使用 `before_request` 和 `after_request` 钩子记录请求信息。 2. 利用 `session` 或自定义 token 标识用户会话。 3. 在内存中（如全局字典）或使用 Redis 存储活跃会话信息。 4. 设置会话超时时间（例如 5 分钟无活动则视为离线）。 5. 提供一个管理接口 `/admin/clients` 查看实时客户端信息。 --- ### ✅ 示例代码（完整可运行） ```python from flask import Flask, session, request, jsonify, g import uuid import time from datetime import datetime, timedelta app = Flask(__name__) app.secret_key = 'your-secret-key-here' # 模拟数据库：存储活跃客户端信息（生产环境建议用 Redis） # 结构: { session_id: { ip, user_agent, last_seen, pages_visited, start_time } } active_clients = {} # 设置会话超时时间（秒） SESSION_TIMEOUT = 300 # 5分钟 @app.before_request def track_client(): # 为新客户端创建唯一 session_id（基于 cookie） if 'client_id' not in session: session['client_id'] = str(uuid.uuid4()) session['start_time'] = time.time() client_id = session['client_id'] now = datetime.now() # 更新或创建客户端记录 if client_id not in active_clients: active_clients[client_id] = { 'ip': request.remote_addr, 'user_agent': request.headers.get('User-Agent', 'Unknown'), 'start_time': now, 'last_seen': now, 'pages_visited': [], 'request_count': 0 } # 记录访问的页面 page_info = { 'url': request.url, 'method': request.method, 'timestamp': now } active_clients[client_id]['pages_visited'].append(page_info) active_clients[client_id]['last_seen'] = now active_clients[client_id]['request_count'] += 1 # 将客户端数据挂载到 g，便于其他函数使用 g.client_data = active_clients[client_id] @app.after_request def remove_inactive(response): # 可选：清理过期会话（也可以用定时任务） cleanup_expired_clients() return response def cleanup_expired_clients(): """清理超过 SESSION_TIMEOUT 未活动的客户端""" now = datetime.now() expired = [] for client_id, data in active_clients.items(): if (now - data['last_seen']).total_seconds() > SESSION_TIMEOUT: expired.append(client_id) for client_id in expired: del active_clients[client_id] @app.route('/') def index(): return '<h1>主页</h1><p><a href="/page1">去页面1</a> | <a href="/admin/clients">查看客户端信息</a></p>' @app.route('/page1') def page1(): return '<h1>这是页面1</h1><p><a href="/">返回主页</a></p>' @app.route('/admin/clients') def admin_clients(): # 清理过期会话 cleanup_expired_clients() clients_list = [] for cid, data in active_clients.items(): duration = (datetime.now() - data['start_time']).total_seconds() clients_list.append({ 'client_id': cid[:8] + '...', 'ip': data['ip'], 'user_agent': data['user_agent'][:50] + '...' if len(data['user_agent']) > 50 else data['user_agent'], 'duration': f"{duration:.1f} 秒", 'last_seen': data['last_seen'].strftime('%H:%M:%S'), 'page_count': len(data['pages_visited']), 'request_count': data['request_count'] }) html = """ <h2>实时客户端信息</h2> <p><strong>当前活跃客户端数量：</strong>{}<br> <strong>刷新此页查看更新</strong></p> <table border="1" cellpadding="8" style="border-collapse: collapse; width: 100%;"> <tr> <th>Client ID</th> <th>IP 地址</th> <th>浏览器</th> <th>持续时间</th> <th>最后活动</th> <th>访问页数</th> </tr> """.format(len(clients_list)) for c in clients_list: html += f""" <tr> <td>{c['client_id']}</td> <td>{c['ip']}</td> <td>{c['user_agent']}</td> <td>{c['duration']}</td> <td>{c['last_seen']}</td> <td>{c['page_count']}</td> </tr> """ html += "</table>" return html # 可选：提供 JSON 接口供前端动态刷新 @app.route('/api/clients') def api_clients(): cleanup_expired_clients() return jsonify({ 'total_active': len(active_clients), 'clients': [ { 'ip': data['ip'], 'pages': len(data['pages_visited']), 'last_seen': data['last_seen'].isoformat(), 'duration': (datetime.now() - data['start_time']).total_seconds() } for data in active_clients.values() ] }) if __name__ == '__main__': app.run(debug=True, host='0.0.0.0', port=5000) ``` --- ### 🔍 代码解释 | 组件 | 功能 | |------|------| | `before_request` | 在每次请求前执行，用于识别或创建客户端标识并记录访问行为。 | | `session['client_id']` | 使用 Flask session 为每个浏览器分配唯一 ID（基于 cookie）。 | | `active_clients` 字典 | 内存中保存所有活跃客户端的信息，包括 IP、UA、访问路径等。 | | `cleanup_expired_clients()` | 清除长时间无活动的客户端，模拟“断开连接”。 | | `/admin/clients` | 管理页面，展示当前活跃客户端列表。 | | `/api/clients` | 返回 JSON 数据，可用于前端轮询显示实时数据。 | --- ### ⚠️ 注意事项 1. **仅限单进程模式有效**：`active_clients` 是内存变量，多 worker（如 Gunicorn）会导致数据不同步。 - ✅ 生产环境建议改用 **Redis** 存储活跃会话。 2. **不能精确反映 WebSocket 连接**：如果需要真正实时通信，请结合 Flask-SocketIO。 3. **隐私问题**：记录 IP 和 User-Agent 可能涉及 GDPR 等合规要求，需告知用户。 --- ### ✅ 增强建议（生产可用） #### 使用 Redis 改造（推荐） ```python import redis import json r = redis.Redis(host='localhost', port=6379, db=0) # 存储格式: SET client:<id> = JSON, EX = 300 def save_client(client_id, data): r.setex(f"client:{client_id}", 300, json.dumps(data, default=str)) def get_all_active_clients(): keys = r.keys("client:*") clients = {} for k in keys: data = r.get(k) if data: clients[k.decode()] = json.loads(data) return clients ``` 然后替换原来的 `active_clients` 字典操作。 --- ###