在Python Flask项目中实现实时Socket通信，最便捷的方案是使用Flask-SocketIO扩展，它封装了WebSocket协议，同时提供了降级兼容方案，能够让Flask应用轻松支持双向实时通信。下面我们将从环境准备到完整示例逐步讲解实现过程。
环境准备与安装
首先需要安装必要的依赖包，Flask-SocketIO需要配合对应的异步服务使用，这里选择常用的eventlet作为异步服务：
pip install flask flask-socketio eventlet
基础实时应用搭建
我们从一个最简单的实时消息收发示例开始，实现客户端连接、发送消息、接收消息的基础流程。
服务端代码实现
服务端需要初始化Flask和SocketIO实例，然后定义对应的事件处理函数：
from flask import Flask, render_template
from flask_socketio import SocketIO, emit
# 初始化Flask应用
app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret_key'
# 初始化SocketIO，指定异步模式为eventlet
socketio = SocketIO(app, async_mode='eventlet')
# 处理客户端连接事件
@socketio.on('connect')
def handle_connect():
print('客户端已连接')
emit('server_response', {'data': '成功连接到服务器'})
# 处理客户端发送的消息事件
@socketio.on('client_message')
def handle_client_message(message):
print(f'收到客户端消息：{message}')
# 向发送消息的客户端返回响应
emit('server_response', {'data': f'服务器已收到你的消息：{message.get("content")}'})
# 处理客户端断开连接事件
@socketio.on('disconnect')
def handle_disconnect():
print('客户端已断开连接')
if __name__ == '__main__':
# 使用socketio运行应用，而不是原生的app.run
socketio.run(app, host='0.0.0.0', port=5000, debug=True)
客户端代码实现
客户端使用原生的JavaScript SocketIO客户端库，实现连接服务器、发送消息、接收响应的功能：
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<title>Flask-SocketIO实时通信示例</title>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
</head>
<body>
<h3>实时消息通信</h3>
<input type="text" id="messageInput" placeholder="输入要发送的消息">
<button onclick="sendMessage()">发送消息</button>
<div id="messageContainer"></div>
<script>
// 连接SocketIO服务器
const socket = io.connect('http://127.0.0.1:5000');
// 监听服务器连接成功响应
socket.on('server_response', function(data) {
const container = document.getElementById('messageContainer');
const p = document.createElement('p');
p.innerText = `服务器响应：${data.data}`;
container.appendChild(p);
});
// 发送消息到服务器
function sendMessage() {
const input = document.getElementById('messageInput');
const content = input.value;
if (content.trim() === '') return;
// 发送client_message事件，携带消息内容
socket.emit('client_message', {content: content});
input.value = '';
}
</script>
</body>
</html>
将上述HTML代码保存为index.html，放在Flask项目的templates目录下，然后添加路由返回页面：
@app.route('/')
def index():
return render_template('index.html')
进阶功能：房间机制
在很多实时场景中，需要将用户分组，比如聊天室的不同房间，Flask-SocketIO提供了房间机制，可以将连接的客户端加入指定房间，实现定向消息推送。
服务端房间相关代码
from flask_socketio import join_room, leave_room
# 处理加入房间事件
@socketio.on('join_room')
def handle_join_room(data):
room_id = data.get('room_id')
join_room(room_id)
emit('server_response', {'data': f'成功加入房间{room_id}'}, room=room_id)
# 处理房间内发送消息事件
@socketio.on('room_message')
def handle_room_message(data):
room_id = data.get('room_id')
content = data.get('content')
# 向指定房间内的所有客户端发送消息
emit('server_response', {'data': f'房间{room_id}消息：{content}'}, room=room_id)
客户端房间操作示例
// 加入房间
function joinRoom() {
const roomId = document.getElementById('roomInput').value;
socket.emit('join_room', {room_id: roomId});
}
// 向房间内发送消息
function sendRoomMessage() {
const roomId = document.getElementById('roomInput').value;
const content = document.getElementById('roomMessageInput').value;
socket.emit('room_message', {room_id: roomId, content: content});
}
注意事项
- 运行应用时必须使用
socketio.run()而不是原生的app.run()，否则SocketIO功能无法正常工作。 - 生产环境建议更换更稳定的异步模式，比如gevent，同时做好异常处理和连接保活机制。
- 如果需要跨域通信，可以在初始化SocketIO时添加
cors_allowed_origins参数配置允许的源。
FlaskFlask_SocketIOSocket通信实时应用修改时间：2026-06-10 09:48:29