Python WebSockets 庫從基礎到實戰(zhàn)使用舉例
更新時間：2025年08月18日 16:05:13 作者：蕭鼎
WebSocket是一種全雙工、持久化的網絡通信協(xié)議,適用于需要低延遲的應用,如實時聊天、股票行情推送、在線協(xié)作、多人游戲等,本文給大家介紹Python WebSockets庫從基礎到實戰(zhàn)使用舉例,感興趣的朋友一起看看吧
1. 引言
WebSocket 是一種全雙工、持久化的網絡通信協(xié)議，適用于需要低延遲的應用，如實時聊天、股票行情推送、在線協(xié)作、多人游戲等。相比傳統(tǒng)的 HTTP 輪詢方式，WebSocket 減少了帶寬開銷，提高了實時性。
在 Python 中，最流行的 WebSocket 庫是 websockets，它是一個基于 asyncio 的輕量級 WebSocket 庫，支持 WebSocket 服務器和客戶端實現(xiàn)。本文將深入介紹 WebSockets 及其在 Python 中的使用方法。
2. 為什么使用 WebSocket？
在傳統(tǒng)的 HTTP 輪詢（Polling）或長輪詢（Long Polling）中，客戶端需要不斷向服務器發(fā)送請求，即使沒有數據更新，也會浪費帶寬和資源。WebSocket 通過單次握手建立持久連接，服務器可以主動推送數據，極大地提高了通信效率。
WebSocket 的優(yōu)勢：
- 低延遲：基于 TCP 連接，減少握手和數據傳輸時間。
- 雙向通信：服務器可以主動向客戶端推送消息，而無需等待請求。
- 減少帶寬消耗：避免 HTTP 頭部的額外開銷，提高吞吐量。
- 適用于實時應用：如聊天、直播、股票行情等。
3. 安裝 WebSockets 庫
首先，我們需要安裝 websockets：
pip install websockets
websockets 依賴 Python 3.6 及以上版本，并且基于 asyncio，所以所有 WebSocket 代碼都是**異步（async）**的。
4. 使用 WebSockets 搭建 WebSocket 服務器
WebSocket 服務器的基本實現(xiàn)只需幾行代碼。
4.1 WebSocket 服務器示例
import asyncio
import websockets
async def echo(websocket, path):
async for message in websocket:
print(f"收到消息: {message}")
await websocket.send(f"服務器響應: {message}")
# 啟動 WebSocket 服務器
start_server = websockets.serve(echo, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()說明：
websockets.serve(echo, "0.0.0.0", 8765)啟動一個 WebSocket 服務器，監(jiān)聽8765端口。async for message in websocket監(jiān)聽客戶端發(fā)送的消息，并在收到后回顯給客戶端。
5. WebSocket 客戶端
WebSocket 客戶端的實現(xiàn)也非常簡單：
import asyncio
import websockets
async def client():
async with websockets.connect("ws://localhost:8765") as websocket:
await websocket.send("Hello, WebSocket Server")
response = await websocket.recv()
print(f"服務器響應: {response}")
asyncio.run(client())說明：
websockets.connect("ws://localhost:8765")連接 WebSocket 服務器。await websocket.send("Hello, WebSocket Server")發(fā)送數據。await websocket.recv()接收服務器的消息。
6. 處理多個客戶端
通常，我們需要處理多個客戶端同時連接。在 WebSockets 中，可以使用 asyncio.gather() 來管理多個 WebSocket 連接。
6.1 廣播消息給所有連接的客戶端
import asyncio
import websockets
connected_clients = set() # 記錄已連接的客戶端
async def handler(websocket, path):
connected_clients.add(websocket)
try:
async for message in websocket:
print(f"收到消息: {message}")
# 廣播給所有客戶端
await asyncio.gather(*(client.send(f"廣播消息: {message}") for client in connected_clients))
finally:
connected_clients.remove(websocket)
start_server = websockets.serve(handler, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()說明：
- 使用
connected_clients集合存儲所有連接的客戶端。 - 在
async for message in websocket內部，遍歷connected_clients，將消息發(fā)送給所有客戶端。
7. WebSocket 服務器的異常處理
實際應用中，客戶端可能會斷開連接，或者發(fā)送非法數據。我們需要在服務器端增加異常處理，以確保服務不會崩潰。
import asyncio
import websockets
async def handler(websocket, path):
try:
async for message in websocket:
print(f"收到: {message}")
await websocket.send(f"服務器回復: {message}")
except websockets.exceptions.ConnectionClosedError:
print("客戶端連接關閉")
except Exception as e:
print(f"發(fā)生錯誤: {e}")
start_server = websockets.serve(handler, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()8. 使用 WebSockets 傳輸 JSON 數據
在 WebSockets 通信中，通常需要傳輸結構化數據，例如 JSON。
服務器端：
import asyncio
import websockets
import json
async def handler(websocket, path):
async for message in websocket:
data = json.loads(message)
response = {"message": f"收到: {data['content']}"}
await websocket.send(json.dumps(response))
start_server = websockets.serve(handler, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()客戶端：
import asyncio
import websockets
import json
async def client():
async with websockets.connect("ws://localhost:8765") as websocket:
data = json.dumps({"content": "Hello, Server"})
await websocket.send(data)
response = await websocket.recv()
print(f"服務器響應: {json.loads(response)}")
asyncio.run(client())9. WebSockets vs. HTTP
| 特性 | WebSockets | HTTP |
|---|---|---|
| 連接方式 | 持久連接 | 請求-響應 |
| 數據推送 | 服務器主動推送 | 需要輪詢 |
| 適用場景 | 實時應用（聊天、直播） | 普通 Web API |
10. WebSocket 實戰(zhàn)：實時聊天室
import asyncio
import websockets
clients = set()
async def chat(websocket, path):
clients.add(websocket)
try:
async for message in websocket:
await asyncio.gather(*(client.send(message) for client in clients))
finally:
clients.remove(websocket)
start_server = websockets.serve(chat, "0.0.0.0", 8765)
asyncio.get_event_loop().run_until_complete(start_server)
asyncio.get_event_loop().run_forever()客戶端可以連接服務器并發(fā)送消息，服務器會廣播給所有連接的用戶，形成一個實時聊天室。
總結
- WebSocket 提供了低延遲、全雙工通信，適用于實時應用。
websockets庫基于asyncio，支持高并發(fā)通信。- WebSockets 可用于聊天系統(tǒng)、股票行情推送、多人協(xié)作、遠程控制等應用場景。
通過本教程，你應該掌握了 Python websockets 庫的使用方法，并能在項目中實現(xiàn)高效的實時通信！??
相關文章
這篇文章主要介紹了Python PO設計模式的具體使用，文中通過示例代碼介紹的非常詳細，對大家的學習或者工作具有一定的參考學習價值，需要的朋友們下面隨著小編來一起學習學習吧2019-08-08
Python實戰(zhàn)之Elasticsearch的高級實現(xiàn)詳解
Elasticsearch是一個功能強大的開源搜索引擎,廣泛應用于各種場景,本文將深入探討如何使用Python與Elasticsearch進行高級實現(xiàn),需要的可以參考下2024-04-04
這篇文章主要介紹了PyTorch兩種安裝方法,本文給大家介紹的非常詳細，對大家的學習或工作具有一定的參考借鑒價值，需要的朋友可以參考下2021-03-03
PyQt5 designer 頁面點擊按鈕跳轉頁面實現(xiàn)
本文主要介紹了PyQt5 designer 頁面點擊按鈕跳轉頁面實現(xiàn)，文中通過示例代碼介紹的非常詳細，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01
這篇文章主要介紹了一些在?Python?編程中可能被忽視的核心功能，包括默認參數、海象運算符、*args?和?**kwargs?的使用等，需要的可參考下2023-08-08
這篇文章主要介紹了python數據結構之二叉樹的遞歸遍歷實例,需要的朋友可以參考下2014-04-04
這篇文章主要介紹了如何用python 批量操作redis數據庫，幫助大家更好的理解和學習使用python，感興趣的朋友可以了解下2021-03-03
這篇文章主要為大家介紹了Python深度學習中pytorch神經網絡多輸入多輸出通道的詳解有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步2021-10-10
這篇文章主要介紹了Django實戰(zhàn)之用戶認證（初始配置），小編覺得挺不錯的，現(xiàn)在分享給大家，也給大家做個參考。一起跟隨小編過來看看吧2018-07-07
在編寫?Python?程序時,日志記錄是一個非常重要的部分,日志不僅幫助你在開發(fā)過程中調試代碼,還可以在程序正式運行時提供診斷信息,幫助定位問題,在本文中,我們將介紹?Python?的?logging?模塊,幫助你了解如何使用這個模塊來記錄和管理日志,需要的朋友可以參考下2024-10-10