Flask + WebSocket 实战：从零构建实时聊天应用（附完整代码）

1. 为什么选择FlaskWebSocket做实时聊天应用实时聊天应用的核心需求是双向即时通信传统HTTP协议每次请求都需要重新建立连接就像每次打电话都要重新拨号一样低效。而WebSocket就像保持通话状态的对讲机建立连接后双方可以随时发送数据。实测下来用Python的Flask框架配合WebSocket技术栈30分钟就能搭建出可用的聊天服务。我最初尝试用轮询Polling实现消息推送发现服务器压力大且延迟高。后来切换到WebSocket方案资源消耗直接降低80%。Flask作为轻量级框架配合Flask-SocketIO这个神器既保留了Python的简洁语法又能处理高并发连接。去年做智能家居项目时就用这套方案实现了设备状态实时同步。2. 环境搭建与基础配置2.1 准备Python环境推荐使用Python 3.8版本太新的版本可能会遇到依赖冲突。先用虚拟环境隔离项目python -m venv chat_env source chat_env/bin/activate # Linux/Mac chat_env\Scripts\activate # Windows安装核心依赖时要注意版本兼容性这是我验证过的稳定组合pip install flask2.0.3 flask-socketio5.3.4 python-engineio4.3.4 python-socketio5.7.22.2 最小化Flask应用结构创建基础项目目录/chat_app ├── app.py # 主程序 ├── static/ # 静态资源 │ └── js/app.js # 前端逻辑 └── templates/ # HTML模板 └── index.html在app.py中初始化应用from flask import Flask, render_template from flask_socketio import SocketIO app Flask(__name__) app.config[SECRET_KEY] your_secret_key_here # 生产环境要用更安全的密钥 socketio SocketIO(app, cors_allowed_origins*) # 允许跨域 app.route(/) def index(): return render_template(index.html) if __name__ __main__: socketio.run(app, debugTrue)3. 实现双向通信核心逻辑3.1 服务端事件处理Flask-SocketIO通过事件驱动模型工作主要处理三类事件连接事件客户端建立连接时触发自定义事件前端主动发起的交互断开事件连接关闭时清理资源socketio.on(connect) def handle_connect(): print(f客户端 {request.sid} 已连接) emit(server_response, {data: 连接成功}) socketio.on(disconnect) def handle_disconnect(): print(f客户端 {request.sid} 已断开) socketio.on(client_message) def handle_message(data): print(f收到消息: {data}) emit(server_broadcast, {user: data[user], msg: data[msg]}, broadcastTrue)3.2 前端交互实现在static/js/app.js中建立WebSocket连接const socket io(); // 连接成功回调 socket.on(connect, () { console.log(已连接到服务器); }); // 接收广播消息 socket.on(server_broadcast, (data) { const chatBox document.getElementById(chat-box); chatBox.innerHTML pb${data.user}:/b ${data.msg}/p; }); // 发送消息函数 function sendMessage() { const user document.getElementById(user).value; const msg document.getElementById(message).value; socket.emit(client_message, {user, msg}); }对应的HTML模板示例div idchat-box styleheight:300px;overflow-y:scroll/div input typetext iduser placeholder你的名字 input typetext idmessage placeholder输入消息 button onclicksendMessage()发送/button script srchttps://cdnjs.cloudflare.com/ajax/libs/socket.io/4.4.1/socket.io.min.js/script script src{{ url_for(static, filenamejs/app.js) }}/script4. 生产环境部署要点4.1 性能优化配置开发时用内置服务器没问题但生产环境需要调整socketio.run(app, host0.0.0.0, port5000, debugFalse, allow_unsafe_werkzeugTrue, keyfilekey.pem, # HTTPS证书 certfilecert.pem)建议配合GunicornEventlet提升并发能力pip install gunicorn eventlet gunicorn -k eventlet -w 4 app:app4.2 常见问题解决方案消息延迟问题开启WebSocket压缩减少传输量socketio SocketIO(app, engineio_loggerTrue, compression_threshold1024)跨域问题除了设置cors_allowed_origins还需要处理预检请求app.after_request def add_cors_headers(response): response.headers.add(Access-Control-Allow-Headers, Content-Type) return responseNginx配置要点在/etc/nginx/sites-available/your_site中添加location /socket.io { proxy_pass http://127.0.0.1:5000; proxy_http_version 1.1; proxy_set_header Upgrade $http_upgrade; proxy_set_header Connection upgrade; }5. 功能扩展实战5.1 实现用户在线列表服务端维护连接字典active_users {} socketio.on(register) def handle_register(username): active_users[request.sid] username emit(user_update, list(active_users.values()), broadcastTrue)前端实时显示socket.on(user_update, (users) { const userList document.getElementById(user-list); userList.innerHTML users.map(u li${u}/li).join(); });5.2 消息持久化存储集成SQLAlchemy保存聊天记录from flask_sqlalchemy import SQLAlchemy db SQLAlchemy(app) class Message(db.Model): id db.Column(db.Integer, primary_keyTrue) user db.Column(db.String(80)) content db.Column(db.Text) timestamp db.Column(db.DateTime, defaultdatetime.utcnow) socketio.on(client_message) def handle_message(data): msg Message(userdata[user], contentdata[msg]) db.session.add(msg) db.session.commit()6. 完整代码示例服务端最终版本# app.py from datetime import datetime from flask import Flask, render_template, request from flask_socketio import SocketIO, emit from flask_sqlalchemy import SQLAlchemy app Flask(__name__) app.config[SECRET_KEY] your-secret-key app.config[SQLALCHEMY_DATABASE_URI] sqlite:///chat.db db SQLAlchemy(app) socketio SocketIO(app, cors_allowed_origins*) class Message(db.Model): id db.Column(db.Integer, primary_keyTrue) user db.Column(db.String(80)) content db.Column(db.Text) timestamp db.Column(db.DateTime, defaultdatetime.utcnow) active_users {} app.route(/) def index(): return render_template(index.html) socketio.on(connect) def handle_connect(): emit(server_response, {data: Connected}) socketio.on(register) def handle_register(username): active_users[request.sid] username emit(user_update, list(active_users.values()), broadcastTrue) socketio.on(client_message) def handle_message(data): msg Message(userdata[user], contentdata[msg]) db.session.add(msg) db.session.commit() emit(server_broadcast, data, broadcastTrue) socketio.on(disconnect) def handle_disconnect(): if request.sid in active_users: del active_users[request.sid] emit(user_update, list(active_users.values()), broadcastTrue) if __name__ __main__: with app.app_context(): db.create_all() socketio.run(app, debugTrue)前端完整实现!-- templates/index.html -- !DOCTYPE html html head title实时聊天室/title style #chat-box { height: 400px; overflow-y: scroll; border: 1px solid #ccc; padding: 10px; } #user-list { float: right; width: 150px; border: 1px solid #eee; padding: 10px; } /style /head body div iduser-list h3在线用户/h3 ul idusers/ul /div div idchat-box/div input typetext idusername placeholder设置昵称 required input typetext idmessage placeholder输入消息 button onclicksendMessage()发送/button script srchttps://cdnjs.cloudflare.com/ajax/libs/socket.io/4.4.1/socket.io.min.js/script script const socket io(); let currentUser ; socket.on(connect, () { const username prompt(请输入昵称) || 匿名用户; document.getElementById(username).value username; currentUser username; socket.emit(register, username); }); socket.on(server_broadcast, (data) { const chatBox document.getElementById(chat-box); const time new Date().toLocaleTimeString(); chatBox.innerHTML p[${time}] b${data.user}:/b ${data.msg}/p; chatBox.scrollTop chatBox.scrollHeight; }); socket.on(user_update, (users) { document.getElementById(users).innerHTML users.map(u li${u}/li).join(); }); function sendMessage() { const msg document.getElementById(message).value; if (msg.trim()) { socket.emit(client_message, { user: currentUser, msg: msg }); document.getElementById(message).value ; } } document.getElementById(message).addEventListener(keypress, (e) { if (e.key Enter) sendMessage(); }); /script /body /html7. 进阶优化方向消息已读回执在消息对象中添加read_by字段当客户端收到消息后发送确认事件socketio.on(message_read) def handle_read(msg_id): msg Message.query.get(msg_id) if msg: msg.read_by current_user db.session.commit()离线消息处理用户重连时检查未读消息socketio.on(reconnect) def handle_reconnect(): unread Message.query.filter( Message.timestamp last_online_time, Message.user ! current_user ).all() for msg in unread: emit(new_message, {id: msg.id, content: msg.content})消息加密传输使用PyNaCl进行端到端加密from nacl.secret import SecretBox key b32-byte-key-for-AES256-encryption box SecretBox(key) encrypted box.encrypt(bsecret message) plaintext box.decrypt(encrypted)