在 Web 开发中，实时显示在线人数是提升用户活跃度感知的经典功能。无论是社交平台、电商网站还是内容社区，这个功能都能让用户直观感受到平台的热度。本文将介绍如何使用 Python Web 技术栈实现这一功能，结合轻量级框架和实时通信技术，打造高效稳定的在线人数统计系统。
一、核心原理：如何追踪用户在线状态？
实现在线人数统计的核心是准确判断 "用户是否在线"，主要涉及两个关键技术点：
用户身份识别：通过会话（Session）或令牌（Token）区分不同用户，确保每个用户只被统计一次
实时状态同步：服务器需要将在线人数变化实时推送给所有客户端，避免页面刷新
本文采用 "Flask 框架 + Redis 存储 + WebSocket 通信" 方案：
Flask 提供轻量级 Web 服务
Redis 存储在线用户信息（支持高并发和过期清理）
WebSocket 实现服务器与客户端的双向实时通信
二、技术栈选择
为平衡开发效率和性能，选用以下技术组合：
后端：Python 3.8+、Flask 2.0+、Flask-SocketIO（WebSocket 封装）
存储：Redis（缓存在线用户，支持自动过期）
前端：HTML5、原生 JavaScript、Bootstrap（样式美化）
部署：Gunicorn（WSGI 服务器）、Eventlet（异步 worker）
三、实战实现：分步构建
1. 环境准备
首先安装所需依赖：
bash
pipinstallflask flask-socketio redis eventlet
确保本地已安装 Redis 并启动（默认端口 6379），Windows 用户可从Redis 官网下载安装包。
2. 后端实现：核心逻辑
创建app.py作为主程序，实现用户跟踪、在线人数统计和 WebSocket 通信：
python
fromflaskimportFlask,render_templatefromflask_socketioimportSocketIO,emitimportredisimportuuidfromdatetimeimporttimedelta# 初始化Flask应用app=Flask(__name__)app.config['SECRET_KEY']='your-secret-key-here'# 生产环境需更换为随机密钥# 初始化SocketIO（使用eventlet作为异步引擎）socketio=SocketIO(app,cors_allowed_origins="*",async_mode='eventlet')# 连接Redis（默认本地配置，生产环境需修改为实际Redis地址）r=redis.Redis(host='localhost',port=6379,db=0)ONLINE_USERS_KEY='online_users'# Redis中存储在线用户的集合键名SESSION_EXPIRE=300# 会话过期时间（秒），5分钟未活动则视为离线defget_online_count():"""获取当前在线人数"""returnr.scard(ONLINE_USERS_KEY)@socketio.on('connect')defhandle_connect():"""处理客户端连接事件"""# 为新连接生成唯一用户ID（实际项目可结合用户登录状态）user_id=str(uuid.uuid4())# 将用户ID存入Redis集合，并设置过期时间< href="https://zq.zhaopin.com/moment/67735435">< href="https://zq.zhaopin.com/moment/67735432">< href="https://zq.zhaopin.com/moment/67735431">< href="https://zq.zhaopin.com/moment/67735401">r.sadd(ONLINE_USERS_KEY,user_id)r.expire(ONLINE_USERS_KEY,SESSION_EXPIRE)# 延长集合过期时间# 向当前连接的客户端发送用户ID（用于后续身份验证）emit('user_connected',{'user_id':user_id})# 广播在线人数更新给所有客户端online_count=get_online_count()emit('online_count_update',{'count':online_count},broadcast=True)print(f"用户 {user_id} 上线，当前在线：{online_count}")@socketio.on('disconnect')defhandle_disconnect():"""处理客户端断开连接事件"""# 从请求上下文获取用户ID（需客户端在连接后发送身份信息）# 实际项目中应通过认证机制获取可靠的用户标识user_id=Noneif'user_id'inrequest.args:user_id=request.args['user_id']ifuser_id:# 从Redis集合中移除用户IDr.srem(ONLINE_USERS_KEY,user_id)online_count=get_online_count()# 广播在线人数更新emit('online_count_update',{'count':online_count},broadcast=True)print(f"用户 {user_id} 下线，当前在线：{online_count}")@socketio.on('heartbeat')defhandle_heartbeat(data):"""处理客户端心跳包，延长会话有效期"""user_id=data.get('user_id')ifuser_idandr.sismember(ONLINE_USERS_KEY,user_id):# 延长用户会话过期时间r.expire(ONLINE_USERS_KEY,SESSION_EXPIRE)print(f"收到用户 {user_id} 心跳，延长会话有效期")@app.route('/')defindex():"""首页路由，返回在线人数统计页面"""returnrender_template('index.html')if__name__=='__main__':# 使用SocketIO运行应用（生产环境需用Gunicorn部署）socketio.run(app,host='0.0.0.0',port=5000,debug=True)
3. 前端实现：实时展示页面
在项目根目录创建templates文件夹，新建index.html：
html
<!DOCTYPEhtml><htmllang="zh-CN"><head><metacharset="UTF-8"><title>Python Web在线人数统计</title><!-- 引入Bootstrap样式 --><linkhref="https://cdn.jsdelivr.net/npm/bootstrap@5.3.0/dist/css/bootstrap.min.css"rel="stylesheet"><style>.container{margin-top:5rem;text-align:center;}.count-display{font-size:6rem;font-weight:bold;color:#2563eb;margin:2rem0;transition:all0.5sease;}.count-display.change{transform:scale(1.1);color:#1d4ed8;}.status-text{color:#64748b;font-size:1.2rem;}</style></head><body><divclass="container"><divclass="card shadow-lg p-5 mx-auto"style="max-width:600px;"><h2class="text-primary">当前在线人数</h2><divid="onlineCount"class="count-display">0</div><pclass="status-text">数据实时更新 · 基于WebSocket技术</p></div></div><!-- 引入Socket.IO客户端库 --><scriptsrc="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script><script>// 连接WebSocket服务器constsocket=io.connect('http://'+document.domain+':'+location.port);// 存储当前用户IDletuserId=null;// 处理服务器返回的用户< href="https://zq.zhaopin.com/moment/67735382">< href="https://zq.zhaopin.com/moment/67735350">< href="https://zq.zhaopin.com/moment/67735343">< href="https://zq.zhaopin.com/moment/67735340">IDsocket.on('user_connected',function(data){userId=data.user_id;console.log('已连接，用户ID：',userId);// 定时发送心跳包（每30秒一次）setInterval(()=>{socket.emit('heartbeat',{user_id:userId});},30000);});// 处理在线人数更新socket.on('online_count_update',function(data){constcountElement=document.getElementById('onlineCount');// 添加数字变化动画countElement.classList.add('change');setTimeout(()=>{countElement.classList.remove('change');},500);// 更新显示countElement.textContent=data.count;});// 处理连接错误socket.on('connect_error',function(error){console.error('连接错误：',error);document.querySelector('.status-text').textContent='连接异常，正在重试...';});// 页面关闭时断开连接window.addEventListener('beforeunload',function(){socket.disconnect();});</script></body></html>
4. 部署配置
为确保生产环境稳定性，创建wsgi.py用于 Gunicorn 部署：
python
# wsgi.pyfromappimportapp,socketioif__name__=='__main__':socketio.run(app,host='0.0.0.0',port=5000)
启动命令（使用 eventlet 作为 worker）：
bash
gunicorn --worker-class eventlet-w1-b0.0.0.0:5000 wsgi:app
四、功能测试
启动 Redis 服务：redis-server
运行应用：python app.py
访问http://localhost:5000，初始显示在线人数 1
打开多个浏览器窗口或使用不同设备访问，人数会实时增加
关闭某个窗口，在线人数会立即减少（WebSocket 断开触发）
测试超时场景：保持窗口打开但不操作，5 分钟后会被视为离线
五、优化与扩展
性能优化：
对于高并发场景，可使用 Redis 集群存储在线用户
增加用户身份验证（结合 Flask-Login），避免匿名用户重复统计
限制 WebSocket 连接频率，防止恶意攻击
功能扩展：
实现用户在线列表，显示用户名和登录时间
添加历史在线峰值统计，通过 Redis 有序集合存储
区分游客和登录用户，分别统计数量：
python
# 存储登录用户r.sadd('online_users:registered',user_id)# 存储游客r.sadd('online_users:guest',user_id)
可靠性提升：
实现客户端自动重连机制，处理网络波动
添加日志记录，追踪用户上线 / 下线事件
定期清理 Redis 中的无效数据（尽管已设置过期时间）
六、注意事项
Redis 配置：生产环境需设置密码认证，并限制访问 IP
跨域问题：部署时需正确配置cors_allowed_origins，避免安全风险
服务器资源：WebSocket 长连接会占用更多内存，需根据服务器配置调整最大连接数
浏览器兼容性：主流浏览器均支持 WebSocket，但可添加轮询降级方案（Flask-SocketIO 已支持）
七、总结
本文使用 Python 的 Flask 框架结合 WebSocket 和 Redis，实现了高效的在线人数实时统计功能。核心逻辑是通过 WebSocket 追踪用户连接状态，用 Redis 存储在线用户信息并自动管理过期数据。
这种方案兼顾了开发效率和运行性能，适合中小规模网站使用。对于大型分布式系统，可进一步结合消息队列（如 RabbitMQ）实现跨服务的在线状态同步，满足更高的并发需求。
