このガイドでは、Python で WebSocket プロキシ サーバーを作成する方法を説明します。
サーバーが行うことは次のとおりです。
- クライアント ID の検証: クライアントの接続を許可する前に、それぞれに固有の「ユーザー キー (API キー)」があるかどうかがチェックされます。
- 別の WebSocket に接続します: サーバーは別の WebSocket サーバーに接続します。
- リレーメッセージ: サーバーは、接続された WebSocket からメッセージを受信し、検証されたすべてのクライアントに送信します。
始める前に:
- Python 3.6 以降のバージョンがインストールされていることを確認してください。 WebSocket には Python 3.6 以降が必要です。
- WebSockets ライブラリをインストールします: ターミナルで次のコマンドを使用してインストールできます。
pip install websockets
1. はじめに
- プロジェクト用に新しいフォルダーを作成します。
- フォルダー内に新しい Python ファイルを作成し、「websocket_proxy_server.py」という名前を付けます。このファイルにはサーバーのすべてのコードが含まれます。
2. WebSocketサーバーの作成
- 必要なライブラリをインポートします。前にインストールしたライブラリが必要になります。
- サーバーの基本構造を構築します。 WebSocket ライブラリを使用して、サーバーの基盤を作成します。
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. クライアントの接続と検証
- サーバーがクライアントからの接続を受け入れるようにすべて設定されていることを確認してください。
- 各クライアントの身元を確認するためのチェックを追加します。クライアントが接続を試行すると、サーバーは「ユーザー キー」を要求する必要があります。正しいキーを持つクライアントのみが接続を許可されます。
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. ソースに接続してメッセージを共有する
- サーバーを元の WebSocket に接続し続ける関数を作成します。
- この関数は、元の 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. サーバーを起動します
- サーバーを起動して接続をリッスンする関数を作成します。
- この関数を実行するコードを追加して、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()
要約すれば
Python ベースの WebSocket プロキシ サーバーの開発に成功しました。このサーバーは、クライアント ID を認証し、指定されたデータ ソースへの永続的な接続を維持し、ソースから受信したメッセージを検証されたすべてのクライアントに効果的に配布できます。この機能は、単一の発信元から多様なユーザー ベースにデータを安全かつ瞬時に配布する必要があるアプリケーションにとって非常に貴重であることがわかります。
次のステップ
最適なパフォーマンスと信頼性を確保するには、徹底的なサーバー テストが不可欠です。接続とメッセージ送信が適切に処理されることを検証します。効率を高めるには、負荷分散メカニズムの実装と接続ヘッダーのカスタマイズを検討してください。最後に、長期間のネットワーク接続に対応するように特別に設計されたクラウド サービスなど、実稼働展開に適した環境にサーバーを展開することをお勧めします。
また、弊社 Web サイトで最初に公開されたチュートリアル: Python プロキシを使用した外国為替 WebSocket のスケーリング
もご覧ください。以上がPython プロキシを使用したスケーラブルな外国為替 WebSocket の実装の詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。

この記事では、バージョン3.10で導入されたPythonの新しい「マッチ」ステートメントについて説明します。これは、他の言語のスイッチステートメントに相当するものです。コードの読みやすさを向上させ、従来のif-elif-elよりもパフォーマンスの利点を提供します

Python 3.11の例外グループは、複数の例外を同時に処理することで、同時シナリオと複雑な操作でエラー管理を改善します。

Pythonの関数注釈は、タイプチェック、ドキュメント、およびIDEサポートの関数にメタデータを追加します。それらはコードの読みやすさ、メンテナンスを強化し、API開発、データサイエンス、ライブラリの作成において重要です。

この記事では、Pythonの単体テスト、その利点、およびそれらを効果的に書く方法について説明します。テスト用のUnittestやPytestなどのツールを強調しています。

記事では、Pythonのアクセス仕様について説明します。Pythonは、厳格な執行ではなく、クラスメンバーの可視性を示すために命名規則を使用します。

記事では、Pythonの\ _ \ _ init \ _ \ _()メソッドと、オブジェクト属性の初期化における自己の役割について説明します。 \ _ \ _ init \ _ \ _()に対するその他のクラス方法と継承の影響についてもカバーされています。

この記事では、@ClassMethod、@StaticMethod、およびPythonのインスタンスメソッドの違いについて説明し、そのプロパティ、ユースケース、および利点を詳述します。必要な機能とDAに基づいて適切な方法タイプを選択する方法を説明します

inpython、youappendelementStoalistusingtheappend()method.1)useappend()forsingleelements:my_list.append(4).2)useextend()or = formultipleElements:my_list.extend(another_list)ormy_list = [4,5,6] .3)forspecificpositions:my_list.insert(1,5).beaware


ホットAIツール

Undresser.AI Undress
リアルなヌード写真を作成する AI 搭載アプリ

AI Clothes Remover
写真から衣服を削除するオンライン AI ツール。

Undress AI Tool
脱衣画像を無料で

Clothoff.io
AI衣類リムーバー

Video Face Swap
完全無料の AI 顔交換ツールを使用して、あらゆるビデオの顔を簡単に交換できます。

人気の記事

ホットツール

EditPlus 中国語クラック版
サイズが小さく、構文の強調表示、コード プロンプト機能はサポートされていません

MinGW - Minimalist GNU for Windows
このプロジェクトは osdn.net/projects/mingw に移行中です。引き続きそこでフォローしていただけます。 MinGW: GNU Compiler Collection (GCC) のネイティブ Windows ポートであり、ネイティブ Windows アプリケーションを構築するための自由に配布可能なインポート ライブラリとヘッダー ファイルであり、C99 機能をサポートする MSVC ランタイムの拡張機能が含まれています。すべての MinGW ソフトウェアは 64 ビット Windows プラットフォームで実行できます。

DVWA
Damn Vulnerable Web App (DVWA) は、非常に脆弱な PHP/MySQL Web アプリケーションです。その主な目的は、セキュリティ専門家が法的環境でスキルとツールをテストするのに役立ち、Web 開発者が Web アプリケーションを保護するプロセスをより深く理解できるようにし、教師/生徒が教室環境で Web アプリケーションを教え/学習できるようにすることです。安全。 DVWA の目標は、シンプルでわかりやすいインターフェイスを通じて、さまざまな難易度で最も一般的な Web 脆弱性のいくつかを実践することです。このソフトウェアは、

MantisBT
Mantis は、製品の欠陥追跡を支援するために設計された、導入が簡単な Web ベースの欠陥追跡ツールです。 PHP、MySQL、Web サーバーが必要です。デモおよびホスティング サービスをチェックしてください。

SecLists
SecLists は、セキュリティ テスターの究極の相棒です。これは、セキュリティ評価中に頻繁に使用されるさまざまな種類のリストを 1 か所にまとめたものです。 SecLists は、セキュリティ テスターが必要とする可能性のあるすべてのリストを便利に提供することで、セキュリティ テストをより効率的かつ生産的にするのに役立ちます。リストの種類には、ユーザー名、パスワード、URL、ファジング ペイロード、機密データ パターン、Web シェルなどが含まれます。テスターはこのリポジトリを新しいテスト マシンにプルするだけで、必要なあらゆる種類のリストにアクセスできるようになります。

ホットトピック









