编程学习网 > 编程语言 > Python > websocket 在python中的使用教程
2023
09-27

websocket 在python中的使用教程

清华大佬耗费三个月吐血整理的几百G的资源，免费分享！....>>>

前段时间测试的时候，需要mock一些数据，用到了websocket，下面通过一个简单的案例介绍websocket的使用。


安装包
pip install bottle
pip install bottle-websocket 
服务器端编写
# websocketdemo.py
from bottle import route, run, static_file
from bottle.ext.websocket import GeventWebSocketServer
from bottle.ext.websocket import websocket
import time

@route("/")
def callback():
    return static_file("websocket.html", root=".")

@route("/websocket", apply=[websocket])
def callback(ws):
    count = 0
    while True:
        count += 1
        time.sleep(1)
        ws.send(str(count))

run(host='localhost', port=9000, server=GeventWebSocketServer)
前端代码编写
将 websocket.html文件放在与websocketdemo.py同目录下，内容为：

<!DOCTYPE html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <title>WebSocket Example</title>
    <style>
      * {
        box-sizing: border-box;
      }

      body {
        font-family: Arial, sans-serif;
        margin: 0;
        padding: 0;
      }

      h1 {
        text-align: center;
      }

      div#output {
          text-align: center;
          font-size: 100px;
        max-width: 600px;
        margin: 0 auto;
        padding: 20px;
        background-color: #f5f5f5;
        border-radius: 5px;
        box-shadow: 0px 2px 4px rgba(0, 0, 0, 0.1);
      }
    </style>
  </head>
  <body>
    <h1>WebSocket Example</h1>
    <div id="output"></div>

    <script>
      const socket = new WebSocket('ws://localhost:9000/websocket');

      socket.addEventListener('open', event => {
        console.log('WebSocket connected');

        function displayMessage(message) {
          const outputDiv = document.getElementById('output');
          outputDiv.innerHTML += message + '<br>';
        }

        socket.addEventListener('message', message => {
          displayMessage(message.data);
        });
      });

      socket.addEventListener('close', event => {
        console.log('WebSocket disconnected');
      });

      socket.addEventListener('error', event => {
        console.error('WebSocket error:', event);
      });
    </script>
  </body>
</html>

上面的代码借助了 「讯飞的星火认知模型」，生成速度还蛮快的。

异步的案例
python提供了一个高级库websockets来实现websocket。

❝
https://websockets.readthedocs.io/en/stable/

❞
pip install websockets
#server.py
import asyncio
import websockets

async def echo(websocket, path):
    #fetch msg
    async for message in websocket:
        print("got a message:{}".format(message))
        await websocket.send(message)

async def main():
    # start a websocket server
    async with websockets.serve(echo, "localhost", 8765):
        await asyncio.Future()  # run forever

asyncio.run(main())
# client.py
import asyncio
import websockets

async def hello():
    async with websockets.connect("ws://localhost:8765") as websocket:
        await websocket.send("Hello world!")
        await websocket.recv()

asyncio.run(hello())

有时候我们的请求是不同路径传来的，那么这个怎么写呢？这里介绍一种新的库：websockets-routes.

❝
https://pypi.org/project/websockets-routes/

❞
pip install websockets-routes
# 2023/5/8 22:46 , @File: server.py, @Comment : python3.9.7
import asyncio
import websockets
import websockets_routes

# 初始化一个router对象
router = websockets_routes.Router()


@router.route("/light/{status}") #添加router的route装饰器，它会路由uri。
async def light_status(websocket, path):
    async for message in websocket:
        print("got a message:{}".format(message))
        print(path.params['status'])
        await asyncio.sleep(2)  # 假装模拟去操作开关灯
        if (path.params['status'] == 'on'):
            await  websocket.send("the light has turned on")
        elif path.params["status"] == 'off':
            await  websocket.send("the light has turned off")
        else:
            await  websocket.send("invalid params")


async def main():
    # rooter是一个装饰器，它的__call__函数有三个参数，第一个参数是self。
    # 所以这里我们需要使用lambda进行一个转换操作，因为serv的wshander函数只能接收2个参数
    async with websockets.serve(lambda x, y: router(x, y), "127.0.0.1", 8765):
        print("======")
        await  asyncio.Future()  # run forever


if __name__ == "__main__":
    asyncio.run(main())

# 2023/5/8 22:46 , @File: client.py, @Comment : python3.9.7
import asyncio
import websockets


async def hello():
    try:
        async with websockets.connect('ws://127.0.0.1:8765/light/on') as websocket:
            light_addr = '00-12-4b-01'
            await websocket.send(light_addr)
            recv_msg = await websocket.recv()
            print(recv_msg)
    except websockets.exceptions.ConnectionClosedError as e:
        print("connection closed error")
    except Exception as e:
        print(e)


if __name__ == '__main__':
    asyncio.run(hello())

建立连接之后，客户端可以随时接收服务器发来的消息。服务器可以依据逻辑，给客户端推送指定消息。

# 2023/5/9 00:35 , @File: server.py, @Comment : python3.9.7
import asyncio
import websockets
import time


async def echo(websocket, path):
    async for message in websocket:
        message = "I got your message: {}".format(message)
        await websocket.send(message)

        while True:
            t = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
            #在服务器回完第一条消息之后，开始轮询时间，当秒数达到0的时候，会主动给客户端回一条消息。
            if str(t).endswith("0"):
                await websocket.send(t)
                # break


asyncio.get_event_loop().run_until_complete(
    websockets.serve(echo, 'localhost', 8765))
asyncio.get_event_loop().run_forever()

# 2023/5/9 00:36 , @File: client.py, @Comment : python3.9.7
import asyncio
import websockets


async def hello(uri):
    async with websockets.connect(uri) as websocket:
        await websocket.send("Hello world!")
        print("< Hello world!")
        while True:
            recv_text = await websocket.recv()
            print("> {}".format(recv_text))


asyncio.get_event_loop().run_until_complete(
    hello('ws://localhost:8765'))

<!DOCTYPE HTML>
<html>
<head>
    <meta charset="utf-8">
    <title>websocket通信客户端</title>
    <script src="https://cdn.staticfile.org/jquery/1.10.2/jquery.min.js"></script>

    <script type="text/javascript">
        function WebSocketTest() {
            if ("WebSocket" in window) {
                // 打开一个 web socket
                var ws = new WebSocket("ws://127.0.0.1:8765");
                // 连接建立后的回调函数
                ws.onopen = function () {
                    // Web Socket 已连接上，使用 send() 方法发送数据
                    ws.send("Hello world!");
                    $("#main").append("<p>" + "<=" + "Hello world!" + "</p>")
                };

                // 接收到服务器消息后的回调函数
                ws.onmessage = function (evt) {
                    var received_msg = evt.data;
                    if (received_msg.indexOf("sorry") == -1) {
                        $("#main").append("<p>" + "=>" + received_msg + "</p>")
                    }
                };
                // 连接关闭后的回调函数
                ws.onclose = function () {
                    // 关闭 websocket
                    alert("连接已关闭...");
                };
            } else {
                // 浏览器不支持 WebSocket
                alert("您的浏览器不支持 WebSocket!");
            }
        }
    </script>
</head>
<body onload="WebSocketTest()">
<div id="main">
</div>
</body>
</html>

效果展示

前面例子的效果展示：


你或许看到了，我没怎么对上面的代码进行解释，因为我也是搜索之后拿来使用，具体原理并不是很清楚，但是能看个大概，可以修改下满足工作的需求。

以上就是“websocket 在python中的使用教程”的详细内容，想要了解更多Python教程欢迎持续关注编程学习网。

扫码二维码 获取免费视频学习资料

Python编程学习

查 看2022高级编程视频教程免费获取