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編程之字符串模板(Template)用法實例分析
這篇文章主要介紹了Python編程之字符串模板(Template)用法,結(jié)合具體實例形式分析了Python字符串模板的功能、定義與使用方法,需要的朋友可以參考下2017-07-07
這篇文章主要為大家詳細(xì)介紹了如何利用Python和PyQt5編寫一個簡單的圖片格式轉(zhuǎn)換器，文中的示例代碼講解詳細(xì)，感興趣的小伙伴可以動手嘗試一下2023-07-07
web.py 十分鐘創(chuàng)建簡易博客實現(xiàn)代碼
web.py是一款輕量級的Python web開發(fā)框架，簡單、高效、學(xué)習(xí)成本低，特別適合作為python web開發(fā)的入門框架2016-04-04
python爬蟲 urllib模塊發(fā)起post請求過程解析
這篇文章主要介紹了python爬蟲 urllib模塊發(fā)起post請求過程解析,文中通過示例代碼介紹的非常詳細(xì)，對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下2019-08-08
本文主要介紹了Tkinter組件Entry的具體使用，Entry組件通常用于獲取用戶的輸入文本，文中通過示例代碼介紹的非常詳細(xì)，具有一定的參考價值，感興趣的小伙伴們可以參考一下2022-01-01
這篇文章主要為大家詳細(xì)介紹了python實現(xiàn)桌面氣泡提示功能，具有一定的參考價值，感興趣的小伙伴們可以參考一下2019-07-07
這篇文章主要介紹了Python中的切片函數(shù)Slice詳解,切片是對有序的集合而言,意思從有序集合里提取數(shù)據(jù)構(gòu)成子集集合,給定提取的起點start、終點end以及方向上的步長step,能否切出非空子集,起點start需能沿步長方向上到達(dá)終點,需要的朋友可以參考下2023-09-09
keras獲得model中某一層的某一個Tensor的輸出維度教程
今天小編就為大家分享一篇keras獲得model中某一層的某一個Tensor的輸出維度教程，具有很好的參考價值，希望對大家有所幫助。一起跟隨小編過來看看吧2020-01-01
這篇文章主要介紹了Python編程把二叉樹打印成多行代碼，具有一定借鑒價值,需要的朋友可以參考下2018-01-01
這篇文章主要介紹了Python中的兩個列表數(shù)值加法,具有很好的參考價值,希望對大家有所幫助,如有錯誤或未考慮完全的地方,望不吝賜教2023-08-08