Flask-SocketIO

Flask-SocketIO使Flask应用程序可以实现客户端和服务器之间的低延迟双向通信。客户端应用程序可以使用 Javascript、Python、C++、Java和Swift中的任何SocketIO客户端库或任何其他兼容客户端来建立与服务器的永久连接。

季布，

5940人浏览 · 2023-08-01 16:06:18

季布， · 2023-08-01 16:06:18 发布

一、简介：

二、安装：

pip install flask-socketio

三、运行Flask-SocketIO要求：

Flask-SocketIO必须要Python 3.6+的版本，可以从下面三个选项中选择：

eventlet是性能最好的选择，支持长轮询和WebSocket传输。
也可以使用gevent。gevent包完全支持长轮询传输，但与eventlet不同的是，gevent没有原生 WebSocket 支持。要添加对 WebSocket 的支持，目前有两个选项。安装gevent-websocket 包为 gevent 添加了 WebSocket 支持，或者可以使用带有 WebSocket 功能的uWSGI Web 服务器。使用 gevent 也是一个性能比较高的选择，但比 eventlet 略低。
也可以使用基于 Werkzeug 的 Flask 开发服务器，但需要注意的是，该 Web 服务器仅用于开发用途，因此它应该仅用于简化开发工作流程，而不用于生产。

Flask-SocketIO会根据安装的包自动检测要使用的异步框架。优先使用 eventlet，其次是 gevent。对于 gevent 中的 WebSocket 支持，首选 uWSGI，其次是 gevent-websocket。如果 eventlet 和 gevent 均未安装，则使用 Flask 开发服务器。

如果使用多个进程，则必须配置消息队列服务以允许服务器协调广播等操作。支持的队列是Redis、RabbitMQ、 Kafka以及Kombu包支持的任何其他消息队列。

在客户端，可以使用官方的 Socket.IO Javascript 客户端库来建立与服务器的连接。还有用 Swift、Java 和 C++ 编写的官方客户端。非官方客户端也可以工作，只要它们实现了 Socket.IO 协议就可以。Python-SocketIO 包实现了SocketIO协议的Python客户端。（注：Python-SocketIO也是Flask-SocketIO作者写的）。

四、初始化

以下代码示例展示了如何将 Flask-SocketIO 添加到 Flask 应用程序：

from flask import Flask, render_template
from flask_socketio import SocketIO

app = Flask(__name__)
app.config['SECRET_KEY'] = 'secret!'
socketio = SocketIO(app)

if __name__ == '__main__':
    socketio.run(app)

当然，也支持init_app()这种方式的初始化风格。请注意 Web 服务器的启动方式，现在变为socketio.run()，而不是app.run()。
当应用程序处于调试模式时，socketio.run()仍然使用的是Werkzeug 开发服务器。在生产模式下，如果可用，则使用 eventlet 网络服务器，否则使用 gevent 网络服务器。如果未安装 eventlet 和 gevent，则使用 Werkzeug 开发 Web 服务器。

Flask 0.11 中引入的命令可以用于启动基于 Werkzeug 的 Flask-SocketIO 开发服务器，但由于缺乏 WebSocket 支持，不推荐这种启动方法。此软件包的先前版本包括该命令的自定义版本，该版本允许在 eventlet 和 gevent 生产服务器上使用 WebSocket，但此功能已停止使用，取而代之的是上面显示的更健壮的socketio.run(app)启动方法。。

应用程序必须向加载 Socket.IO 库并建立连接的客户端提供一个页面：

<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js" integrity="sha512-q/dWJ3kcmjBLU4Qc47E4A9kTB4m3wuTY7vkFJDTZKjTs8jhyGQnaUrxa0Ytd0ssMZhbNua9hE+E7Qv1j+DyZwA==" crossorigin="anonymous"></script>
<script type="text/javascript" charset="utf-8">
    var socket = io();
    socket.on('connect', function() {
        socket.emit('my event', {data: 'I\'m connected!'});
    });
</script>

测试案例

from flask import Flask, request,render_template
from flask_socketio import SocketIO

app = Flask(__name__)
socketio = SocketIO(app)


@app.route('/')
def hello_world():  # put application's code here
    return render_template("index.html")


@socketio.on('connect')
def connect():
    print('ip:' + request.remote_addr)
    print('sid:' + request.sid)

@socketio.on('zhiliao')
def zhiliao_event(data):
    print(data)
    return {"result":"success"}

if __name__ == '__main__':
    socketio.run(app)

<!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <title>SocketIO测试</title>
</head>
<body>
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"
        integrity="sha512-q/dWJ3kcmjBLU4Qc47E4A9kTB4m3wuTY7vkFJDTZKjTs8jhyGQnaUrxa0Ytd0ssMZhbNua9hE+E7Qv1j+DyZwA=="
        crossorigin="anonymous"></script>

<script type="text/javascript" charset="utf-8">
    var socket = io();
    socket.on('connect', function () {
        socket.emit('my event', {data: 'I\'m connected!'});
    });
    socket.emit("zhiliao",{"message":"i am dd"},function (result){
        console.log("时间发送成功");
        console.log(result);
    })

</script>

</body>
</html>