websockets 是一個用於在 Python 中構建WebSocket服務器和客戶端的庫,側重於正確性、簡單性、健壯性和性能。
它支持多種網絡 I/O 和控制流範例:
默認實現基於asyncioPython 的標準異步 I/O 框架。它提供了一個優雅的基於協程的 API。它非常適合同時處理許多客戶端的服務器。
該threading實現對於客戶來說是一個很好的選擇,尤其是在您不熟悉asyncio. 它也可以用於不需要為許多客戶端提供服務的服務器。
Sans -I/O實現旨在集成到第三方庫中,通常是應用程序服務器,此外還供 websockets 內部使用。
這是一個帶有 API 的回顯服務器asyncio:
#!/usr/bin/env python
import asyncio
from websockets.server import serve
async def echo(websocket):
async for message in websocket:
await websocket.send(message)
async def main():
async with serve(echo, "localhost", 8765):
await asyncio.Future() # run forever
asyncio.run(main())
以下是客戶端如何使用threadingAPI 發送和接收消息:
#!/usr/bin/env python
import asyncio
from websockets.sync.client import connect
def hello():
with connect("ws://localhost:8765") as websocket:
websocket.send("Hello world!")
message = websocket.recv()
print(f"Received: {message}")
hello()
不要擔心打開和關閉握手、ping 和 pong 或 WebSocket 規範中描述的任何其他行為。websockets 會在後台處理這個問題,因此您可以專注於您的應用程序!
此外,websockets 提供了一個交互式客戶端:
python -m websockets ws://localhost:8765/
Connected to ws://localhost:8765/.
> Hello world!
< Hello world!
Connection closed: 1000 (OK).