python用websockets创建服务端websocket创建客户端-Toy模板网

这篇具有很好参考价值的文章主要介绍了python用websockets创建服务端websocket创建客户端。希望对大家有所帮助。如果存在错误或未考虑完全的地方，请大家不吝赐教，您也可以点击"举报违法"按钮提交疑问。

服务端

import asyncio
import json
import base64
import time
import cv2
import numpy as np
import websockets


class WebServer:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.start_server = self.connect()
        self.clients = []

    async def echo(self, websocket, path):
        self.clients.append(websocket)
        client_ip, client_port = websocket.remote_address
        await websocket.send(json.dumps({"type": "handshake"}))

        while True:
            try:
                recv_text = await websocket.recv()
                with open("aa.txt", "w") as f:
                    f.write(recv_text)
                print(recv_text )
                msg = "收到消息"
                await websocket.send(msg)
                await asyncio.sleep(0.2)

            except websockets.ConnectionClosed:
                print("ConnectionClosed...", path)  # 链接断开
                self.clients.remove(websocket)
                break
            except websockets.InvalidState:
                print("InvalidState...")  # 无效状态
                self.clients.remove(websocket)
                break
            except Exception as e:
                print(e)
                # self.clients.remove(websocket)
                # break

    async def send_msg(self, msg, websocket):
        print('send msg:', msg)
        if websocket is not None:
            await websocket.send(msg)
        await asyncio.sleep(0.2)

    async def broadcast_msg(self, msg):
        for user in self.clients:
            await user.send(msg)

    def connect(self):
        start_server = websockets.serve(self.echo, self.host, self.port)
        print("连接成功！")
        return start_server

    def run(self):
        print("已启动！")
        asyncio.get_event_loop().run_until_complete(self.start_server)
        asyncio.get_event_loop().run_forever()


if __name__ == '__main__':
    ws = WebServer("192.168.6.28", 8000)
    ws.run()

客户端文章来源地址https://www.toymoban.com/news/detail-835664.html

import json
import websocket


class WebClient:
    def __init__(self, host, port):
        self.host = host
        self.port = port
        self.conn = None
        self.flag = False

    def connect(self):
        try:
            url = f"ws://{self.host}:{self.port}"
            self.conn = websocket.create_connection(url)
            self.flag = True
        except Exception as err:
            self.flag = False

    def close(self):
        self.conn.close()

    def recv(self):
        data = self.conn.recv(1024)
        print(data)

    def send(self, data):
        self.conn.send(data)


if __name__ == '__main__':
    host = "192.168.6.28"
    # host = "127.0.0.1"
    port = 8000
    ws = WebClient(host, port)
    if not ws.flag:
        ws.connect()
    with open("bb.txt") as f:
        data = f.read()
    ws.send(data)

到了这里，关于python用websockets创建服务端websocket创建客户端的文章就介绍完了。如果您还想了解更多内容，请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章，希望大家以后多多支持TOY模板网！