服務(wù)端

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...")  # 無效狀態(tài)
                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()

客戶端文章來源地址http://www.zghlxwxcb.cn/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)

到了這里，關(guān)于python用websockets創(chuàng)建服務(wù)端websocket創(chuàng)建客戶端的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關(guān)文章，希望大家以后多多支持TOY模板網(wǎng)！