Flask 中如何使用 Websocket 构建 Web实时应用程序

Flask 是用 Python 构建中小型网络应用程序的微型框架的绝佳选择。

Websockets 是一种在服务器和客户端之间同时交换数据的强大技术，可以部署到需要实时更新数据的动态网络应用程序中。

在本文中，我们将在 Flask 中使用 Websockets，以了解其工作原理并释放 Websockets 的强大功能！在 HTTP 连接中，客户端（浏览器）向服务器发送请求，等待响应，收到响应后关闭连接。如果客户端需要进一步的数据（或更多数据），则需要再次发出新的请求（连接）。这也被称为半双工连接，一次只能有一方发送/接收数据。另一方面，Websockets 是一种全双工连接，客户端和服务器可以同时交换数据。由于服务器和客户端之间的连接是开放的（长期的），一旦有数据，就会立即发送给客户端，因此客户端不需要每次单独请求数据。

好了，理论说到这里就够了，让我们深入构建一个Web应用程序来演示 Websockets 的魔力。我们将创建一个Web应用程序，在终端上使用 Linux ping 测试来检查互联网连接（使用公共 DNS 服务器 8.8.8.8），如下所示。

[test@flask ~]$ ping -c 5 8.8.8.8
PING 8.8.8.8 (8.8.8.8) 56(84) bytes of data.
64 bytes from 8.8.8.8: icmp_seq=1 ttl=117 time=12.7 ms
64 bytes from 8.8.8.8: icmp_seq=2 ttl=117 time=12.8 ms
64 bytes from 8.8.8.8: icmp_seq=3 ttl=117 time=12.9 ms
64 bytes from 8.8.8.8: icmp_seq=4 ttl=117 time=15.1 ms
64 bytes from 8.8.8.8: icmp_seq=5 ttl=117 time=12.4 ms

让我们使用 Websockets 将其发布到 Flask Web应用程序中。

创建 Python 虚拟环境以安装 Flask

[test@flask ~]$ /usr/local/bin/python3.8 -m venv myvirtual
[test@flask ~]$ source myvirtual/bin/activate
(myvirtual) [test@flask ~]$ pip install Flask

安装 Flask-SocketIO

我们可以使用 Flask-SocketIO 在 Flask 中使用 websockets：

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

(myvirtual) [test@flask ~]$ pip install flask-socketio

创建 Flask 应用程序

在虚拟环境中创建一个存放应用程序的目录。让我们创建一个名为 “flaskapp “的目录，并在其中创建一个名为 “myapp.py “的文件，用于存放应用程序代码。同时，在 “flaskapp “中创建一个名为 “templates “的目录。在 “templates “中创建一个名为 “base.html “的文件。整个 Flask 目录结构应如下所示：

/home/test/flaskapp
| — myapp.py
| — templates/
|— — — base.html

现在，在 “myapp.py “中输入以下代码。这是您通过 websockets 向浏览器发布 ping 结果的应用代码（与我们在 Linux 终端运行 ping 命令时看到的代码相同）。

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

app = Flask(__name__)
socketio = SocketIO(app,debug=True,cors_allowed_origins='*',async_mode='eventlet')


@app.route('/home')
def main():
        return render_template('base.html')

@socketio.on("my_event")
def checkping():
    for x in range(5):
        cmd = 'ping -c 1 8.8.8.8|head -2|tail -1'
        listing1 = subprocess.run(cmd,stdout=subprocess.PIPE,text=True,shell=True)
        sid = request.sid
        emit('server', {"data1":x, "data":listing1.stdout}, room=sid)
        socketio.sleep(1)

在上面的代码中，首先从上面安装的 flask 和 socketio 软件包中导入了必要的对象，然后创建了在应用程序中使用的实例。请注意，将 async_mode 定义为 “eventlet”。根据 SocketIO 文档，需要使用描述的异步服务之一。我们使用了 eventlet 并安装了它。

接下来，将 flask 路由（端点）定义为 /home，它将渲染名为 “base.html “的 HTML 页面，稍后将创建这个页面。

接下来定义了一个名为 “my_event “的自定义事件。在上述代码中，服务器从客户端（浏览器）接收到名为 “my_event “的事件后，将运行名为 “checkping “函数的事件处理程序，使用名为 “server “的 socketio 发射事件将 ping 命令的结果发送回客户端（浏览器）。客户端和服务器都可以使用事件。因此，总的来说，使用 for 循环运行了 5 次 ping 命令，并在运行后立即向客户端发送（发射）连续的 O/P 返回结果。

请注意，我们捕获的是 ping 结果（listing1.stdout）和 for 循环执行（x）的输出，分别位于消息 data 和 data1 中（我们可以使用任何名称）。至于为什么要捕获 for 循环的范围（0 至 4），稍后就会明白。实际上，这是我们应用程序的一个附加功能，以避免在数据仍在从服务器发送时多次点击按钮。

使用 Gunicorn 和 Nginx 提供 Flask 应用程序

现在应用程序已准备就绪，可以部署 Web 服务器来为应用程序提供服务。我们将使用非常适合生产用例的 Gunicorn + Nginx 组合。

(myvirtual) [test@flask ~]$ pip install gunicorn

[test@flask ~]$ sudo yum install epel-release
[test@flask ~]$ sudo yum install nginx

修改 nginx.conf 文件，加入以下内容，并确保在 server_name 部分写入服务器的 IP 地址。

server {
    listen       80;
    listen       [::]:80;
    server_name  <IP-Address>;


location / {
 include proxy_params;
 proxy_pass http://127.0.0.1:5000;

}

location /socket.io {
 proxy_pass http://127.0.0.1:5000/socket.io;
 proxy_redirect off;
 proxy_buffering off;
 proxy_set_header Host $host;
 proxy_set_header X-Real-IP $remote_addr;
 proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
 proxy_http_version 1.1;
 proxy_set_header Upgrade $http_upgrade;
 proxy_set_header Connection “Upgrade”;
 proxy_hide_header Access-Control-Allow-Origin;

}

}

另外，在 /etc/nginx/ 下创建一个名为“proxy_params”的文件，包含以下内容：

proxy_set_header Host $http_host;
proxy_set_header X-Real-IP $remote_addr;
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
proxy_set_header X-Forwarded-Proto $scheme;

接下来，为 Flask 应用程序创建 Linux systemd 服务，这样就可以使用 Linux systemctl 命令来停止/启动它。此外，通过这种方法，还可以确保 Flask 应用程序在服务器重启时自动启动。

在 /etc/systemd/system/ 下创建名为 “myapp.service “的文件，内容如下：

[Unit]
Description=Gunicorn instance to serve Flask
After=network.target

[Service]
User=test
WorkingDirectory=/home/test/flaskapp/
ExecStart=/home/test/myvirtual/bin/gunicorn --worker-class eventlet -w 1 myapp:app --bind 0.0.0.0:5000
Restart=always


[Install]
WantedBy=multi-user.target

为 Flask 应用程序创建 HTML 页面

在上述 Flask 应用程序目录结构中，我们创建了 “templates “文件夹，其中包含一个 “base.html “文件。当浏览器指向 http://<server-ip>/home 时，该文件就会呈现。

这是应用程序提供的页面，用于查看服务器发送的 ping 结果。

将以下内容放入 “base.html “文件：

<!DOCTYPE html>
<html lang="en">
  <head>
<meta http-equiv="Cache-control" content="no-cache" charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/3.0.4/socket.io.js" integrity="sha512-aMGMvNYu8Ue4G+fHa359jcPb1u+ytAF+P2SCb+PxrjCdO3n3ZTxJ30zuH39rimUggmTwmh2u7wvQsDTHESnmfQ==" crossorigin="anonymous"></script>
<link href="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/css/bootstrap.min.css" rel="stylesheet" integrity="sha384-EVSTQN3/azprG1Anm3QDgpJLIm9Nao0Yz1ztcQTwFspd3yD65VohhpuuCOmLASjC" crossorigin="anonymous">
<script src="https://cdn.jsdelivr.net/npm/bootstrap@5.0.2/dist/js/bootstrap.bundle.min.js" integrity="sha384-MrcW6ZMFYlzcLA8Nl+NtUVF0sA7MsXsP1UyJoMp4YLEuNSfAP+JcXn/tWtIaxVXM" crossorigin="anonymous"></script>
<script type="text/javascript" src="//code.jquery.com/jquery-1.4.2.min.js"></script>
<style>
.navbar-brand {
font-family: 'Merriweather';
font-size: 30px;
}
.inner {
text-align: center;
margin: auto;
margin-top: 10px;
padding: 10px;
border-style: solid;
width: 50%;
color: black;

}
</style>
</head>

<body>
<nav class="navbar navbar-expand-lg navbar-dark bg-dark">
  <div class="container">
   <a class="navbar-brand" href={{ url_for('main') }}>Flask</a>
    <button class="navbar-toggler" type="button" data-bs-toggle="collapse" data-bs-target="#navbarNavDropdown" aria-controls="navbarNavDropdown" aria-expanded="false" aria-label="Toggle navigation">
      <span class="navbar-toggler-icon"></span>
    </button>
  </div>
</nav>
<div class="container">
<div class="inner">
 <div id="header"></div><br>
<button class="btn btn-primary" id="checkbutton" onClick="myupdate()">Submit</button>
<div id="demo"></div>
</div>
</div>
<script>
        const socket = io(); //socketio connection to server//
        $(document).ready(function() {
 });
socket.on("connect", () => {
 console.log("connected");
        document.getElementById("header").innerHTML = "<h3>" + "Websocket Connected" + "</h3";


});

socket.on("disconnect", () => {
 console.log("disconnected");
        document.getElementById("header").innerHTML = "<h3>" + "Websocket Disconnected" + "</h3>";
});

function myupdate() {
  //Event sent by Client
 socket.emit("my_event", function() {
 });
}

// Event sent by Server//
socket.on("server", function(msg) {
        let myvar = JSON.parse(msg.data1);
        //Check if entire data is sent by server//
        if (myvar == "4") {
                document.getElementById("demo").innerHTML = "";
                document.querySelector('#checkbutton').innerText = "Submit";
                document.getElementById("checkbutton").style.cursor = "pointer";
                document.getElementById("checkbutton").disabled = false;
                document.getElementById("checkbutton").className = "btn btn-primary";
 
        }

        else {
                document.getElementById("demo").innerHTML += msg.data + "<br>";
                document.getElementById("checkbutton").disabled = true;
                document.getElementById("checkbutton").innerHTML = "Loading..";
                document.getElementById("checkbutton").style.cursor = "not-allowed";
                document.getElementById("checkbutton").style.pointerEvents = "auto";
        }
});

</script>
</body>
</html>

上面的 HTML 中最重要的部分是事件的 socketIO 事件处理程序 - connect、disconnect、my_event 和 server。这就是正在使用的 4 个事件。socketIO connect 和 disconnect 事件是用于连接和断开 websocket 的内置事件。另外两个事件–“my_event “和 “server “是自定义事件。my_event “事件用于从客户端向服务器发送数据（不过，在这里我们实际上没有向服务器发送任何数据，而只是用于触发按钮点击事件）。按钮点击会运行一个名为 “myupdate “的函数，该函数会创建名为 “my_event “的事件，用于向服务器发送数据。服务器（Flask 应用程序）会捕获这个 “my_event “事件，并运行事件处理程序（检查函数）来处理 ping 结果，然后使用服务器发送的名为 “server “的事件（可以是任何用户定义的名称）发送结果。客户端捕获 Flask 应用程序发送的名为 “server “的事件，并在网页上显示 ping 结果。

请记住，在上述 Flask 应用程序代码中，我们还将 for 循环的执行情况（循环运行次数）作为 “data1 “发送给客户端。我们正在检查循环运行的次数，以验证循环是否真正结束（从而确认完整的 ping 数据已发送到客户端），这是在第 5 次迭代之后（在 Python For 循环中，迭代次数为 0 到 4）。因此，如果第 4 次迭代完成，就意味着完整的数据已从服务器发送到客户端。我们在上述 HTML 代码中使用此检查来验证是否仍在接收数据。如果仍在接收数据，则禁用按钮点击。如果已收到完整数据，则重新启用按钮。

启动 Flask 应用程序

(myvirtual) [test@flask ~]$ sudo systemctl enable myapp.service 
(myvirtual) [test@flask ~]$ sudo systemctl start myapp.service

打开浏览器，访问 http://<server-ip>/home 查看应用程序，显示如下。尝试停止 flask 应用程序（sudo systemctl stop myapp.service），您应该会在网页上看到 “Websocket 已断开连接 “的信息。

点击 “提交 “按钮，查看直接发送到浏览器的实时 ping 结果！

就是这样！这只是使用 Websockets 构建Web应用的开始。

Flask 中如何使用 Websocket 构建 Web实时应用程序

创建 Python 虚拟环境以安装 Flask

使用 Gunicorn 和 Nginx 提供 Flask 应用程序

为 Flask 应用程序创建 HTML 页面

启动 Flask 应用程序

相关推荐

WebSocket 前世今生 ？建立 WebSocket 的 6 种方式？

Golang 基于 Gorilla WebSocket 开发简单的聊天应用程序

WebRTC 与 WebSocket 哪个好？WebRTC 与 WebSocket 可同时使用吗

.NET 8 中使用 WebSocket 实现实时聊天功能

使用 Laravel 和 WebSockets 构建实时应用程序

Pushpin：利用 WebSockets 优化实时通信

发表回复

WebSocket 前世今生？建立 WebSocket 的 6 种方式？