快捷導(dǎo)航

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

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

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

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

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

1. WebSocket 基礎(chǔ)概念

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

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

WebSocket 握手過程：

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

2. Python WebSocket 庫

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

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

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

3. 安裝 websockets 庫

在使用 websockets 庫之前，我們需要先安裝它。可以通過 pip 來安裝：

pip install websockets

4. WebSocket 服務(wù)器端實(shí)現(xiàn)

創(chuàng)建 WebSocket 服務(wù)器

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

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():
    # 啟動(dòng) WebSocket 服務(wù)器，監(jiān)聽 8765 端口
    server = await websockets.serve(handle_connection, "localhost", 8765)
    print("WebSocket server started on ws://localhost:8765")
    await server.wait_closed()
 
# 啟動(dòng)事件循環(huán)
asyncio.run(main())

代碼說明

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

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

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

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

創(chuàng)建 WebSocket 客戶端

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

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

代碼說明

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

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

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

6. 異常處理

在實(shí)際應(yīng)用中，WebSocket 連接可能會(huì)因?yàn)榫W(wǎng)絡(luò)中斷、服務(wù)器關(guān)閉等原因而失敗。我們可以在客戶端和服務(wù)器端都加入適當(dāng)?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 服務(wù)器端異常處理

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. 廣播消息

如果你想讓服務(wù)器向所有已連接的客戶端廣播消息，可以維護(hù)一個(gè)客戶端列表并發(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. 認(rèn)證和授權(quán)

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

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}")
    # 處理正常的消息