本指南將教您如何使用 Python 建立 WebSocket 代理伺服器。

伺服器將執(zhí)行以下操作：

• 驗證客戶端身分：在允許客戶端連線之前，會檢查每個用戶端是否擁有唯一的「使用者金鑰（API Key）」。
• 連接到另一個 WebSocket： 伺服器將連接到一個單獨的 WebSocket 伺服器。
• 中繼訊息：伺服器將從連接的 WebSocket 接收訊息並將其傳送給所有經(jīng)過驗證的用戶端。

開始之前：

• 確保您已安裝Python 3.6或更高版本。 WebSockets 需要 Python 3.6 或更高版本。
• 安裝 WebSockets 庫：您可以在終端機中使用以下命令來安裝它。

pip install websockets

1. 入門

• 為您的專案建立一個新資料夾。
• 在資料夾中建立一個新的 Python 檔案並將其命名為「websocket_proxy_server.py」。該文件將保存您伺服器的所有代碼。

2. 建立WebSocket伺服器

• 導(dǎo)入所需的庫。您將需要之前安裝的庫。
• 建構(gòu)伺服器的基本結(jié)構(gòu)。使用 WebSockets 庫為您的伺服器建立基礎(chǔ)。

import asyncio
import websockets
import json

class WebSocketProxy:

    def init(self, source_url, symbols):

        self.source_url = source_url
        self.clients = set()
        self.symbols = symbols
        self.valid_user_key = "yourValidUserKey"  # Single valid user key for authentication

    async def on_open(self, ws):

        print("Connected to source")
        symbols_str = ",".join(self.symbols.keys())
        init_message = f"{{"userKey":"your_api_key", "symbol":"{symbols_str}"}}"
        await ws.send(init_message)

3. 連線並驗證客戶端

• 確保伺服器已全部設(shè)定為接受來自客戶端的連線。
• 新增檢查以驗證每個客戶的身份。當(dāng)客戶端嘗試連線時，伺服器應(yīng)要求提供「使用者金鑰」。只有具有正確密鑰的客戶端才被允許連線。

async def client_handler(self, websocket, path):

        try:

            # Wait for a message that should contain the authentication key
            auth_message = await asyncio.wait_for(websocket.recv(), timeout=10)
            auth_data = json.loads(auth_message)
            user_key = auth_data.get("userKey")

            if user_key == self.valid_user_key:
                self.clients.add(websocket)
                print(f"Client authenticated with key: {user_key}")

                try:
                    await websocket.wait_closed()

                finally:
                    self.clients.remove(websocket)

            else:

                print("Authentication failed")
                await websocket.close(reason="Authentication failed")
        except (asyncio.TimeoutError, json.JSONDecodeError, KeyError):
            print("Failed to authenticate")
            await websocket.close(reason="Failed to authenticate")

4. 連接到來源並共享訊息

• 建立一個函數(shù)，使伺服器保持與原始 WebSocket 的連線。
• 此函數(shù)應(yīng)該會自動將從原始 WebSocket 接收到的訊息傳送到所有成功驗證的用戶端。

async def source_handler(self):
        async with websockets.connect(self.source_url) as websocket:
            await self.on_open(websocket)
            async for message in websocket:
                await self.broadcast(message)

    async def broadcast(self, message):
        if self.clients:
            await asyncio.gather(*(client.send(message) for client in self.clients))

5. 啟動伺服器

• 建立一個函數(shù)來啟動伺服器並監(jiān)聽連線。
• 新增程式碼來執(zhí)行此函數(shù)，啟動您的 WebSocket 代理伺服器。

def run(self, host="localhost", port=8765):
        start_server = websockets.serve(self.client_handler, host, port)
        asyncio.get_event_loop().run_until_complete(start_server)
        asyncio.get_event_loop().run_until_complete(self.source_handler())
        asyncio.get_event_loop().run_forever()

if name == "main":
    symbols = {"EURUSD": {}, "GBPUSD": {}, "USDJPY": {}, "AUDUSD": {}, "USDCAD": {}}
    source_url = "ws://example.com/source"
    proxy = WebSocketProxy(source_url, symbols)
    proxy.run()

總之

您已經(jīng)成功開發(fā)了一個基於Python的WebSocket代理伺服器。此伺服器可以驗證客戶端身份，維護(hù)與指定資料來源的持久連接，並將從來源接收的訊息有效地分發(fā)給所有經(jīng)過驗證的客戶端。事實證明，對於需要將資料從單一來源安全即時傳播到不同用戶群的應(yīng)用程式來說，此功能非常寶貴。

下一步

徹底的伺服器測試對於確保最佳效能和可靠性至關(guān)重要。它驗證其對連接和訊息傳輸?shù)恼_處理。為了提高效率，請考慮實施負(fù)載平衡機制和自訂連線標(biāo)頭。最後，建議將伺服器部署到適合生產(chǎn)部署的環(huán)境，例如專門為容納長期網(wǎng)路連線而設(shè)計的雲(yún)端服務(wù)。

另外，請查看我們網(wǎng)站上最初發(fā)布的教學(xué)：使用 Python 代理程式擴展外匯 WebSocket

以上是使用 Python 代理實現(xiàn)可擴展的外匯 WebSocket的詳細(xì)內(nèi)容。更多資訊請關(guān)注PHP中文網(wǎng)其他相關(guān)文章！