本指南將教您如何使用 Python 創(chuàng)建 WebSocket 代理服務(wù)器。

服務(wù)器將執(zhí)行以下操作：

• 驗(yàn)證客戶端身份：在允許客戶端連接之前，會(huì)檢查每個(gè)客戶端是否擁有唯一的“用戶密鑰（API Key）”。
• 連接到另一個(gè) WebSocket： 服務(wù)器將連接到一個(gè)單獨(dú)的 WebSocket 服務(wù)器。
• 中繼消息：服務(wù)器將從連接的 WebSocket 接收消息并將其發(fā)送給所有經(jīng)過驗(yàn)證的客戶端。

開始之前：

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

pip install websockets

1. 入門

• 為您的項(xiàng)目創(chuàng)建一個(gè)新文件夾。
• 在文件夾中創(chuàng)建一個(gè)新的 Python 文件并將其命名為“websocket_proxy_server.py”。該文件將保存您服務(wù)器的所有代碼。

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

• 導(dǎo)入所需的庫(kù)。您將需要之前安裝的庫(kù)。
• 構(gòu)建服務(wù)器的基本結(jié)構(gòu)。使用 WebSockets 庫(kù)為您的服務(wù)器創(chuàng)建基礎(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. 連接并驗(yàn)證客戶端

• 確保服務(wù)器已全部設(shè)置為接受來(lái)自客戶端的連接。
• 添加檢查以驗(yàn)證每個(gè)客戶的身份。當(dāng)客戶端嘗試連接時(shí)，服務(wù)器應(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. 連接到源并共享消息

• 創(chuàng)建一個(gè)函數(shù)，使服務(wù)器保持與原始 WebSocket 的連接。
• 此函數(shù)應(yīng)該自動(dòng)將從原始 WebSocket 接收到的消息發(fā)送到所有成功驗(yàn)證的客戶端。

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. 啟動(dòng)服務(wù)器

• 創(chuàng)建一個(gè)函數(shù)來(lái)啟動(dòng)服務(wù)器并監(jiān)聽連接。
• 添加代碼來(lái)運(yùn)行此函數(shù)，啟動(dòng)您的 WebSocket 代理服務(wù)器。

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ā)了一個(gè)基于Python的WebSocket代理服務(wù)器。該服務(wù)器可以驗(yàn)證客戶端身份，維護(hù)與指定數(shù)據(jù)源的持久連接，并將從源接收到的消息有效地分發(fā)給所有經(jīng)過驗(yàn)證的客戶端。事實(shí)證明，對(duì)于需要將數(shù)據(jù)從單一來(lái)源安全即時(shí)傳播到不同用戶群的應(yīng)用程序來(lái)說(shuō)，此功能非常寶貴。

下一步

徹底的服務(wù)器測(cè)試對(duì)于確保最佳性能和可靠性至關(guān)重要。它驗(yàn)證其對(duì)連接和消息傳輸?shù)恼_處理。為了提高效率，請(qǐng)考慮實(shí)施負(fù)載平衡機(jī)制和自定義連接標(biāo)頭。最后，建議將服務(wù)器部署到適合生產(chǎn)部署的環(huán)境，例如專門為容納長(zhǎng)期網(wǎng)絡(luò)連接而設(shè)計(jì)的云服務(wù)。

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

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