Python中實現(xiàn)WebSocket的示例詳解

更新時間：2025年03月10日 15:59:27 作者：王子良.

WebSocket 是一種計算機通信協(xié)議,它為客戶端和服務器之間的雙向通信提供了一個全雙工的通道,本篇文章將詳解 Python 中如何實現(xiàn) WebSocket,感興趣的可以了解下

WebSocket 是一種計算機通信協(xié)議，它為客戶端和服務器之間的雙向通信提供了一個全雙工的通道。通過 WebSocket，客戶端和服務器可以在一個長期的連接中保持通信，而無需每次都建立新的連接。這個特性在實時聊天、在線游戲、股票行情等應用中非常有用。

本篇文章將詳解 Python 中如何實現(xiàn) WebSocket，內(nèi)容包括 WebSocket 的基礎(chǔ)概念、Python 中的 WebSocket 庫、簡單的客戶端和服務器端代碼示例、以及一些高級使用場景。

1. WebSocket 基礎(chǔ)概念

WebSocket 協(xié)議最初是由 IETF（Internet Engineering Task Force）發(fā)布的，目的是提供一個更高效的通信協(xié)議，可以通過單個長連接在客戶端和服務器之間進行雙向通信。WebSocket 的通信通過在 HTTP 協(xié)議上建立一個持久連接（即 WebSocket 握手），并且支持全雙工通信，即數(shù)據(jù)可以同時從客戶端發(fā)送到服務器，也可以從服務器發(fā)送到客戶端。

WebSocket 與傳統(tǒng)的 HTTP 協(xié)議有很大的不同。HTTP 是一個請求-響應式的協(xié)議，客戶端每次需要獲取數(shù)據(jù)時都需要向服務器發(fā)送請求，而 WebSocket 則是通過一個初始握手請求將通信通道建立起來，之后雙方可以隨時發(fā)送數(shù)據(jù)。

WebSocket 握手過程：

客戶端向服務器發(fā)起一個 HTTP 請求，并要求升級到 WebSocket 協(xié)議。
服務器響應客戶端的請求，確認升級協(xié)議。
客戶端和服務器建立 WebSocket 連接，可以開始全雙工通信。

2. Python WebSocket 庫

在 Python 中，我們通常使用以下幾種庫來實現(xiàn) WebSocket：

websockets：一個輕量級的 Python WebSocket 實現(xiàn)，支持異步編程。
socket.io：一個更為高級的庫，支持事件驅(qū)動和自動重連等特性，常用于實時應用。
asyncio：雖然 asyncio 本身并不是一個 WebSocket 庫，但它為 WebSocket 的異步編程提供了基礎(chǔ)支持。

在本篇文章中，我們將主要介紹 websockets 庫，并使用它來實現(xiàn) WebSocket 客戶端和服務器。

3. 安裝 websockets 庫

在使用 websockets 庫之前，我們需要先安裝它?？梢酝ㄟ^ pip 來安裝：

pip install websockets

4. WebSocket 服務器端實現(xiàn)

創(chuàng)建 WebSocket 服務器

WebSocket 服務器通常運行在一個特定的端口，并等待客戶端的連接請求。我們可以通過 websockets 庫來實現(xiàn)一個簡單的 WebSocket 服務器。

import asyncio
import websockets
 
async def handle_connection(websocket, path):
    # 接收到客戶端的消息
    print(f"Connected to {websocket.remote_address}")
    try:
        async for message in websocket:
            print(f"Received message: {message}")
            # 發(fā)送消息回客戶端
            await websocket.send(f"Server received: {message}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        print(f"Disconnected from {websocket.remote_address}")
 
async def main():
    # 啟動 WebSocket 服務器，監(jiān)聽 8765 端口
    server = await websockets.serve(handle_connection, "localhost", 8765)
    print("WebSocket server started on ws://localhost:8765")
    await server.wait_closed()
 
# 啟動事件循環(huán)
asyncio.run(main())

代碼說明

handle_connection 是處理每個客戶端連接的異步函數(shù)。每當有客戶端連接時，服務器將接收到的消息打印出來，并將消息返回給客戶端。

websockets.serve() 用于啟動 WebSocket 服務器，監(jiān)聽指定的地址和端口（此處為 localhost 和 8765）。

await server.wait_closed() 保證服務器持續(xù)運行，直到手動關(guān)閉。

5. WebSocket 客戶端實現(xiàn)

創(chuàng)建 WebSocket 客戶端

WebSocket 客戶端用于與服務器建立連接，并發(fā)送和接收消息。我們使用 websockets.connect 來連接到服務器。

import asyncio
import websockets
 
async def send_message():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        # 向服務器發(fā)送消息
        await websocket.send("Hello, Server!")
        print("Message sent to server.")
        
        # 接收服務器返回的消息
        response = await websocket.recv()
        print(f"Received from server: {response}")
 
# 啟動事件循環(huán)
asyncio.run(send_message())

代碼說明

websockets.connect(uri) 用于與 WebSocket 服務器建立連接。uri 是服務器的地址，此處為 ws://localhost:8765。

使用 await websocket.send() 向服務器發(fā)送消息。

使用 await websocket.recv() 接收來自服務器的消息。

6. 異常處理

在實際應用中，WebSocket 連接可能會因為網(wǎng)絡中斷、服務器關(guān)閉等原因而失敗。我們可以在客戶端和服務器端都加入適當?shù)漠惓Ｌ幚怼?/p>

6.1 客戶端異常處理

async def send_message():
    uri = "ws://localhost:8765"
    try:
        async with websockets.connect(uri) as websocket:
            await websocket.send("Hello, Server!")
            print("Message sent to server.")
            response = await websocket.recv()
            print(f"Received from server: {response}")
    except websockets.exceptions.ConnectionClosed as e:
        print(f"Connection closed: {e}")
    except Exception as e:
        print(f"Error: {e}")
 
asyncio.run(send_message())

6.2 服務器端異常處理

async def handle_connection(websocket, path):
    print(f"Connected to {websocket.remote_address}")
    try:
        async for message in websocket:
            print(f"Received message: {message}")
            await websocket.send(f"Server received: {message}")
    except websockets.exceptions.ConnectionClosed as e:
        print(f"Connection closed: {e}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        print(f"Disconnected from {websocket.remote_address}")

7. WebSocket 高級使用

7.1. 廣播消息

如果你想讓服務器向所有已連接的客戶端廣播消息，可以維護一個客戶端列表并發(fā)送消息。

clients = set()
 
async def handle_connection(websocket, path):
    clients.add(websocket)
    try:
        async for message in websocket:
            print(f"Received message: {message}")
            # 廣播消息給所有客戶端
            for client in clients:
                if client != websocket:
                    await client.send(f"Broadcast message: {message}")
    except Exception as e:
        print(f"Error: {e}")
    finally:
        clients.remove(websocket)
        print(f"Disconnected from {websocket.remote_address}")

7.2. 認證和授權(quán)

你可以使用 WebSocket 協(xié)議的初始握手階段進行認證和授權(quán)，比如使用 HTTP header 傳遞認證信息。

async def handle_connection(websocket, path):
    headers = websocket.request_headers
    token = headers.get('Authorization')
    if token != "expected_token":
        await websocket.send("Unauthorized")
        await websocket.close()
        return
    print(f"Client authorized with token: {token}")
    # 處理正常的消息