当前位置：首页 > news >正文

flask-socketio相关总结

news 来源：原创 2025/7/13 6:37:05

flask-socketio是一个为flask应用程序添加的实时双向通信功能的扩展库，有了这个库，就可以在flask应用中应用websocket协议，帮助flask实现低延迟、双向的客户端、服务端通信。客户端通过任何SocketIO官方库，都能与服务器建立长连接。

文章目录

1、HTTP与WebSocket区别
2、flask-socketio使用
- 2.1 安装
- 2.2 关于异步服务的依赖
- 2.3 flask-socketio简单使用
- 2.4 服务端接收消息
- 2.5 服务器发送消息
- 2.6 广播
- 2.7 房间
- 2.8 连接事件
- 2.9 部署
- 2.10 跨域问题

1、HTTP与WebSocket区别

HTTP与WebSocket是两种在网络通信中广泛使用的协议，它们各自具有独特的特点和适用场景。

HTTP
定义： HTTP（Hyper Text Transfer Protocol，超文本传输协议）是一个简单的请求-响应协议，通常运行在TCP之上。
特点： HTTP是基于客户/服务器模式，客户与服务器建立连接后，客户向服务器提出请求，服务器接受请求并作出应答，然后客户与服务器关闭连接。这种连接是一次性的，并且是单向无状态的，每次连接只处理一个请求。HTTP协议传输的数据通常是文本或二进制数据。
适用场景： 适用于一次性、不会高频更新的数据传输场景，如网页浏览、图片加载等；且由于是无状态协议，因此也非常适合处理大量并发请求，如Web服务器对多个用户的请求进行处理。
WebSocket
定义： WebSocket是一种在单个TCP连接上进行全双工通信的协议。
特点： WebSocket需要浏览器和服务器握手建立连接。一旦连接建立，客户端和服务器可以双向通信，即服务器可以主动向客户端推送信息，客户端也可以主动向服务器发送信息。且这种连接是有状态的，连接建立后，客户端和服务器之间的连接将保持活动状态，直到被任何一方（客户端或服务器）终止。WebSocket协议可以传输任意格式的数据，包括文本、二进制、JSON等。
适用场景： 适用于需要实时性、高频更新的数据传输场景，如在线聊天室、实时股票行情、在线游戏等。由于WebSocket支持双向通信和持久连接，因此非常适合处理需要实时交互的应用场景。

下图是WebSocket与HTTP协议工作图示区别（图片来自迷途小书童的Note）：
在这里插入图片描述

2、flask-socketio使用

flask-socketio官方文档链接：https://flask-socketio.readthedocs.io/en/latest/

2.1 安装

flask-socketio可通过pip快速安装：
pip install flask-socketio

2.2 关于异步服务的依赖

flask-socketio需要底层异步服务器的支持，而且会自己根据当前环境存在的异步服务自动选择，可供选择的服务框架有三种，顺序为：eventlet > gevent > werkzeug

eventlet，性能最好，支持长轮询和Websocket协议，通过 pip install eventlet 安装。
gevent，它能支持多样设置，gevent支持长轮询方式，但是不支持原生WebSocket，为了能支持原生WebSocket，需要选取如下两种方案，一是通过命令 pip install gevent-websocket 安装 gevent-websocket 库为gevent增加WebSocket支持；二是使用带有WebSocket功能的uWSGI Web服务器。性能方面，gevent表现不错，但不如eventlet。
使用基于werkzeug的flask开发服务器，但需要注意的是，它缺乏其他两个选项的性能，因此只应用于简单的开发环境，而且它也仅支持长轮询传输。

2.3 flask-socketio简单使用

首先创建flask应用实例，然后初始化flask-socketio扩展，接着定义事件处理函数，并使用@socketio.on装饰器来监听特定的事件，最后启动应用时，使用socketio.run()来代替app.run()。代码如下：

from flask import Flask, render_template
from flask_socketio import SocketIO, emit
import timeapp = Flask(__name__)
app.config['SECRET_KEY'] = 'secret'
socketio = SocketIO()
socketio.init_app(app)@app.route('/')
def index():"""demo page"""return render_template('index.html')  # 用于展示逐字打印效果的网页@socketio.on('start_stream')
def start_stream():text = "这是一段要逐步返回的文字"for char in text:emit('new_char', char)time.sleep(1)if __name__ == '__main__':socketio.run(app, port=5002, debug=True)

前端代码必须加载http://Socket.IO库，并建立连接：

<!DOCTYPE html>
<html><head><script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.2/socket.io.min.js"></script>
</head><body><div id="result"></div><script>const socket = io.connect('http://localhost:5002');socket.on('connect', function () {socket.emit('connect success.');});socket.on('new_char', function (char) {if (char) {document.getElementById('result').innerHTML += char;}});</script>
</body></html>

2.4 服务端接收消息

当使用SocketIO时，消息被双方作为事件接收。前端发送消息时，服务端需要为这些事件注册处理程序。

下面是服务端为一些未命名事件的处理程序：

@socketio.on('message')
def handle_message(data):print('received message: ' + data)    # 未命名事件使用字符串消息@socketio.on('json')
def handle_json(json):print('received json: ' + str(json))   # 未命名事件使用json数据

最灵活的事件类型使用自定义事件名称

# 最灵活的事件类型使用自定义事件名称，比如这里的‘my event’。这些事件的消息数据可以是字符串、字节、int或JSON
@socketio.on('my event')
def handle_my_custom_event(json):print('received json: ' + str(json))# 自定义命名事件也可以支持多个参数
@socketio.on('my_event')
def handle_my_custom_event(arg1, arg2, arg3):print('received args: ' + arg1 + arg2 + arg3)

当装饰的函数名符合python命名规则且不与其他python标识符冲突时，可以用socketio.event装饰器

# 当装饰的函数名符合python命名规则且不与其他python标识符冲突时，可以用socketio.event装饰器，这时不需要指定事件的元数据类型
# 名称message、json、connect和disconnect是保留的，因此不能用于命名事件。
@socketio.event
def my_custom_event(arg1, arg2, arg3):print('received args: ' + arg1 + arg2 + arg3)

命名空间，namespace

# flask-socketio还支持命名空间，所以在不同命名空间下，可以定义相同的事件名，它们不冲突。
# 若未指定命名空间，则使用名称为‘/’的默认全局命名空间
@socketio.on('my event', namespace='/test')
def handle_my_custom_namespace_event(json):print('received json: ' + str(json))

若不想使用装饰器，用socketio的on_event方法调用实现消息的处理

# 不用装饰器语法也可以，用socketio的on_event方法调用实现消息的处理
def my_function_handler(data):passsocketio.on_event('my event', my_function_handler, namespace='/test')

客户端如何确认服务器已经收到了它们发的消息

# 从处理函数返回的任何值都将作为回调函数的参数传递给客户端
# 下面例子中，客户端回调函数将被调用两个参数，‘one’和2。如果处理程序函数不返回任何值，则调用客户端回调函数时不带参数。
@socketio.on('my event')
def handle_my_custom_event(json):print('received json: ' + str(json))return 'one', 2

2.5 服务器发送消息

SocketIO事件处理程序可以使用send()和emit()函数向连接的客户端发送消息。

下面是一些简单的例子，将服务端接收到前端发来的消息再发送回去。其中send用于未命名事件，emit用于命名事件。

from flask_socketio import send, emit@socketio.on('message')
def handle_message(message):send(message)@socketio.on('json')
def handle_json(json):send(json, json=True)@socketio.on('my event')
def handle_my_custom_event(json):emit('my response', json)

命名空间(namespace)，Send()和emit()默认使用传入消息的名称空间，当然也可以指定另外不同的命名空间。

@socketio.on('message')
def handle_message(message):send(message, namespace='/chat')@socketio.on('my event')
def handle_my_custom_event(json):emit('my response', json, namespace='/chat')

要用元组包含发送带有多个参数的事件

@socketio.on('my event')
def handle_my_custom_event(json):emit('my response', ('foo', 'bar', json), namespace='/chat')

服务端如何确认客户器已经收到了它们发的消息

def ack():print('message was received!')@socketio.on('my event')
def handle_my_custom_event(json):emit('my response', json, callback=ack)   # 确认回调

2.6 广播

SocketIO的另一个非常有用的特性是消息广播。Flask-SocketIO通过send()和emit()的可选参数broadcast=True来支持此功能。广播功能开启时，所有连接这个命名空间的客户端（包括发送者在内）都会收到这个消息。命名空间未指定时，所有连接全局命名空间的客户端会接收消息。注意，广播消息不会触发回调函数。

@socketio.on('my event')
def handle_my_custom_event(data):emit('my response', data, broadcast=True)

上面的例子都是服务端在接收到客户端发来的消息事件后作出的响应，服务端如何首先给客户端发送消息。

def some_function():socketio.emit('some event', {'data': 42})

这里的socketio.send() 和socketio.emit()方法不同于处在事件函数上下文中的send() 和emit()。另外，由于是在一个普通函数中，没有客户端上下文信息，所以 broadcast=True是默认的，不必指定。

2.7 房间

实际应用场景中，可能需要给用户分组。比如，聊天室，不同用户只能收到他们所在房间的消息。通过join_room() 和 leave_room() 可以实现上述功能：

from flask_socketio import join_room, leave_room@socketio.on('join')
def on_join(data):username = data['username']room = data['room']join_room(room)send(username + ' has entered the room.', to=room)@socketio.on('leave')
def on_leave(data):username = data['username']room = data['room']leave_room(room)send(username + ' has left the room.', to=room)

send()和emit()函数接受to=room用于把消息发送到指定房间。
所有客户端连接时，会被分配一个房间。默认房间名称为连接的session ID，flask中通过request.sid获取该ID。客户端能加入所有存在的房间。客户端断开时，所有它加入的房间都会移除它。上下文外的socketio.send() 和 socketio.emit()也可以接收room参数，来给房间中所有客户端广播。

2.8 连接事件

连接和断开事件用于验证客户端是否有连接权限

@socketio.on('connect')
def test_connect(auth):emit('my response', {'data': 'Connected'})@socketio.on('disconnect')
def test_disconnect():print('Client disconnected')

连接处理程序中的auth参数是可选的。客户端可以使用它来传递字典格式的令牌等身份验证数据。

2.9 部署

内置服务器
最简单的部署方式，就是安装eventlet或gevent，然后调用socketio.run(app)。需要注意的是，socketio.run(app)是用于生产环境的，但前提必须确保eventlet或gevent已经安装；否则只会调用Flask自带的服务器，这个服务器仅限于测试环境使用。
Gunicorn服务器
使用gunicorn作为web服务器，使用eventlet或gevent工作线程。需要安装gunicorn，eventlet或者gevent。
通过gunicorn启动eventlet服务器的命令行：gunicorn --worker-class eventlet -w 1 module:app
通过gunicorn启动gevent服务器的命令行：gunicorn -k gevent -w 1 module:app
当使用gunicorn与gevent，而且选择由geevent-WebSocket提供的WebSocket支持时，命令行如下：
gunicorn -k geventwebsocket.gunicorn.workers.GeventWebSocketWorker -w 1 module:app
gunicorn的第三个选择是使用线程工作器，以及用于WebSocket支持的simple-websocket包。对于CPU占用较多的应用程序可以使用：
gunicorn -w 1 --threads 100 module:app
所有这些命令，module是python的定义在应用实例当中的包或模块，app就是应用实例本身。
uWSGI服务器
当将uWSGI服务器与gevent结合使用时，Socket.IO服务器可以利用uWSGI的原生WebSocket支持，启动命令如下：
uwsgi --http :5000 --gevent 1000 --http-websockets --master --wsgi-file app.py --callable app

2.10 跨域问题

如果是前后端分离的系统中，就会出现跨域问题，在网络应用开发中，跨域资源共享（Cross Origin Resource Sharing，简称CORS）是一种机制，允许服务器与指定的来源或域名之间共享资源。使用CORS，我们可以灵活地控制不同域之间的数据传输，实现安全、可靠的跨域访问。单纯在flask应用中，可以使用flask-cors扩展库来实现CORS功能。

# 安装： pip install flask-cors
from flask import Flask, jsonify
from flask_cors import CORSapp = Flask(__name__)
CORS(app) # 允许应用的所有视图都可以跨域访问....

但是在app用flask-socketio中，上述方法并不起作用了，我们需要在 socketio 初始化的时候加入必要的参数来实现跨域访问：

socketio = SocketIO()
socketio.init_app(app, cors_allowed_origins='*')