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