【websocket】websocket-client 与 websockets-Toy模板网

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

websocket-client

websocket-client 是 websocket 客户端，提供了对ws低级API的访问。通过导入 websocket 库使用，websocket 库是基于事件驱动的设计模式，通过定义回调函数来处理接收到的消息、错误和连接关闭等事件。
优势：

兼容多个 Python 版本，包括 Python 2.7 和 Python 3.x。
简单易用，入门门槛较低。
提供了基本的 WebSocket 功能，可以满足一般需求。

劣势：文章来源地址https://www.toymoban.com/news/detail-654948.html

功能相对较少，不支持一些高级特性，如异步操作和性能优化。

"""用 websocket 创建长连接"""
import time

import websocket
from gevent import thread


def on_message(ws, message):
    # 处理收到的消息
    print("Received: " + message)


def on_error(ws, error):
    # 处理错误
    print("Error: " + str(error))


def on_close(ws):
    # 关闭连接
    print("Connection closed")


def on_open(ws):
    # 连接成功后的操作
    def run(*args):
        # 发送心跳包或其他持续性操作
        while True:
            ws.send("Ping")
            time.sleep(1)  # 每隔一秒发送一次

    thread.start_new_thread(run, ())


if __name__ == "__main__":
    # 创建 WebSocket 连接
    websocket.enableTrace(True)
    ws = websocket.WebSocketApp(
        'ws://your-websocket-url',
        on_message=on_message,
        on_error=on_error,
        on_close=on_close
    )
    ws.on_open = on_open

    # 运行 WebSocket 客户端
    ws.run_forever()

websockets

websockets 具有 server端和 client端，采用异步操作模式，与 asyncio 模块无缝集成，可以实现高性能的 WebSocket 服务器和客户端。

python版本仅支持3.6及更高版本

优势：

高性能和可扩展性
提供了丰富的特性：支持心跳包、自定义协议和 SSL/TLS 加密等高级功能。

劣势：

不兼容 Python 2.x 版本。

"""服务端"""
import asyncio
import websockets


async def handle(ws, path):
    name = await ws.recv()
    print(f"接收: {name}")

    greeting = f"已收到 {name}!"
    await ws.send(greeting)
    print(f"发送: {greeting}")


if __name__ == '__main__':
    s = websockets.serve(handle, "127.0.0.1", 9451)

    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(s)
    event_loop.run_forever()

"""客户端"""
import asyncio
import websockets


async def hello():
    uri = "ws://127.0.0.1:9451"
    async with websockets.connect(uri) as websocket:
        name = input("发送: ")

        await websocket.send(name)
        greeting = await websocket.recv()
        print(f"接收: {greeting}")


if __name__ == '__main__':
    event_loop = asyncio.get_event_loop()
    event_loop.run_until_complete(hello())