Python WebSocket 教程：从零开始

Python 中实现 WebSocket 服务

今天带大家来通过 Python 来实现 WebSocket 服务。

安装 Tornado 库

我们需要使用 pip 进行库的安装：

pip install tornado

Python 实现 WebSocket

我们可以在 Python 中使用 Tornado 进行 WebSocket 服务的实现~

import tornado.ioloop
import tornado.web
import tornado.websocket
import time

class WebSocketHandler(tornado.websocket.WebSocketHandler):
    def open(self):
        print("open success")
        # 定时器，每秒向前端发送一次数据
        self.timer = tornado.ioloop.PeriodicCallback(self.send_data, 1000)
        self.timer.start()

    def on_close(self):
        self.timer.stop()

    def send_data(self):
        # 向前端发送当前时间
        self.write_message('Now is' + str(time.time()))

application = tornado.web.Application([
    (r'/', WebSocketHandler),
])

if __name__ == '__main__':
    application.listen(3001)
    tornado.ioloop.IOLoop.instance().start()

该代码创建了一个名为 WebSocketHandler 的类, 继承 tornado.websocket.WebSocketHandler. 这个类实现了开启，关闭以及消息接收方法。

运行代码

在终端中输入：

python server.py

我们可以看到，WebSocket 服务的地址为：

ws: //localhost:3001

前端页面连接 WebSocket

页面编写

我们需要创建一个 index.html，并写入以下代码：

<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta http-equiv="X-UA-Compatible" content="IE=edge">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <title>Document</title>
  <script>
    window.onload = () => {
        if ('WebSocket' in window) {
            // 创建websocket连接
            let ws = new WebSocket('ws://127.0.0.1:3001/websocket');
            // 成功连接的时候推送一条消息，此时服务端就可以开始推送数据了
            ws.onopen = () => {
                console.log('websocket success---');
                ws.send('success');
            }
            ws.onmessage = (message) => {
                let data = message.data;
                console.log('get websocket message---', data);
            }
            ws.onerror = () => {
                console.error('websocket fail');
            }
        } else {
            console.error('dont support websocket');
        };
    };
</script>

</head>
<body>
  
</body>
</html>

启动页面

在 index.html 右键，并打开此文件：

查看效果

可以看到后端先接收到前端的信息，open 建立成功：

然后建立连接后，后端一直在为前端输出信息。

在 Apifox 中调试 WebSocket

如果你要调试 WebSocket 接口，并确保你的应用程序能够正常工作。这时，一个强大的接口测试工具就会派上用场。

Apifox 是一个比 Postman 更强大的接口测试工具，Apifox = Postman + Swagger + Mock + JMeter。它支持调试 http（s）、WebSocket、Socket、gRPC、Dubbo 、SSE 等多种协议的接口，这使得它成为了一个非常全面的接口测试工具，所以强烈推荐去下载体验！

在 Apifox 中，你需要创建一个 WebSocket 请求以便进入相应的界面进行必要的信息填写。以下是步骤：

新建一个 HTTP 项目后，在项目中点击“+”按钮。
输入 WebSocket 服务的地址，例如 http://localhost:3000。
选择性地填写要传输的消息（Message）和参数（Params）。
填写完必要信息后，点击“保存”按钮以保留本次请求信息，方便下次调用。

立即体验 Apifox

要与服务器进行通信，你首先需要建立一个 WebSocket 连接。只需点击“连接”按钮即可。连接成功后，Apifox 会给出提示，同时服务器也会对此作出响应。

成功连接后，你可以使用 Apifox 提供的“发送”按钮，来发送消息和参数给服务器。

发送消息后，服务器同样可以向客户端回传信息，这一功能展示了 Apifox 的 WebSocket 功能的简便性，快去试试吧！

免费使用 Apifox

总结

本文介绍了如何使用 Python 和 Tornado 库实现 WebSocket 服务，包括安装 Tornado、创建 WebSocketHandler 类、运行服务及前端页面连接等步骤。此外，还推荐了 Apifox 作为调试 WebSocket 接口的强大工具，详细说明了在 Apifox 中创建 WebSocket 请求、建立连接、发送和接收消息的具体操作，使开发者能更高效地测试和开发 WebSocket 应用。