Posted inPython

Pythonのwebsocketsライブラリ入門：双方向リアルタイムアプリケーションの構築ガイド

Python tutorial - IT technology blog
Python tutorial - IT technology blog

クイックスタート：5分で双方向接続を構築する

pip経由でライブラリをインストールします：

pip install websockets

server.pyを作成し、クライアントからのメッセージを受信して即座に応答する処理を記述します：

# server.py
import asyncio
import websockets

async def echo(websocket):
    async for message in websocket:
        print(f"[サーバー受信]: {message}")
        await websocket.send(f"サーバー応答: {message}")

async def main():
    async with websockets.serve(echo, "localhost", 8765):
        print("サーバーが ws://localhost:8765 で待機中...")
        await asyncio.Future()  # バックグラウンドプロセスを維持

if __name__ == "__main__":
    asyncio.run(main())

続いて、テストデータを送信するclient.pyを作成します：

# client.py
import asyncio
import websockets

async def send_message():
    uri = "ws://localhost:8765"
    async with websockets.connect(uri) as websocket:
        await websocket.send("クライアントからのPing!")
        response = await websocket.recv()
        print(f"[クライアント受信]: {response}")

if __name__ == "__main__":
    asyncio.run(send_message())

2つのターミナルウィンドウを開き、まずpython server.pyを実行してからpython client.pyを実行します。5ms未満の低レイテンシで、ほぼ瞬時に結果が画面に出力されます。

HTTPポーリング vs WebSocket：切り替えの判断基準

従来のHTTPはステートレスなRequest-Responseモデルで動作します。状態を継続的に更新するには、クライアントが1〜3秒ごとにリクエストを定期送信するポーリング技術を用いる必要がありました。

このアプローチは負荷が高まるにつれて大きな課題が顕在化します：

  • HTTPオーバーヘッド： 各リクエストには500〜1000バイトのHTTPヘッダー（Cookie、User-Agent、Authなど）が含まれます。一方WebSocketはハンドシェイク完了後、わずか2〜6バイトのフレーミングペイロードで済みます。
  • サーバーリソース消費： 1万人のユーザーが毎秒ポーリングすると、サーバーには毎秒1万リクエストが発生します。WebSocketなら1万本の確立済みTCPコネクションを維持し、イベント発生時のみデータを転送します。
  • レイテンシ（遅延）： ポーリングではポーリング間隔分の遅延が常に生じます。WebSocketならイベント発生と同時にクライアントへデータをプッシュ可能です。

websocketsライブラリの裏側にあるAsyncioの仕組み

本ライブラリは完全にasyncioのイベントループ上で動作します。ある接続がネットワークI/O待ちの間、CPUは即座に別の接続からのパケット処理へと切り替わります。このノンブロッキングI/Oにより、シングルスレッドのPythonワーカーであっても極めて少ないメモリ消費で数万のソケットを同時に維持できます。

応用：複数ユーザー向けブロードキャストルームの構築

株価ボードやチャットルームのように、多数のサブスクライバーへ同時にデータを送信するユースケースは非常に一般的です。

ライブラリには、Set内の全ソケットに対してメッセージを最適化して一斉送信するwebsockets.broadcast関数が標準で用意されています：

# chat_server.py
import asyncio
import websockets

CONNECTED_CLIENTS = set()

async def handler(websocket):
    CONNECTED_CLIENTS.add(websocket)
    client_ip = websocket.remote_address
    print(f"+ クライアント接続: {client_ip} (オンライン数: {len(CONNECTED_CLIENTS)})")
    
    try:
        async for message in websocket:
            # アクティブな全クライアントへブロードキャスト
            websockets.broadcast(CONNECTED_CLIENTS, f"{client_ip[0]}: {message}")
    except websockets.exceptions.ConnectionClosed:
        pass
    finally:
        CONNECTED_CLIENTS.remove(websocket)
        print(f"- クライアント切断: {client_ip} (オンライン数: {len(CONNECTED_CLIENTS)})")

async def main():
    async with websockets.serve(handler, "0.0.0.0", 8765):
        print("ブロードキャストサーバーがポート 8765 で起動しました...")
        await asyncio.Future()

if __name__ == "__main__":
    asyncio.run(main())

本番環境デプロイ時の重要ポイント3選

1. ロードバランサー経由の接続維持（ハートビート / Ping-Pong）

AWS ALBやCloudflareなどは、アイドル状態が60秒を超えるとTCPコネクションを切断する傾向があります。定期的なPingを設定して接続を常にキープアライブしましょう：

# 20秒ごとにPingを送信、10秒間Pong応答がない場合は切断
async with websockets.serve(
    handler, 
    "0.0.0.0", 
    8765, 
    ping_interval=20, 
    ping_timeout=10
):
    await asyncio.Future()

2. クライアント側の自動再接続（指数バックオフ）

モバイル回線やWi-Fi環境は頻繁に不安定になります。クライアント側では例外をキャッチし、サーバーへの過負荷を防ぐため指数バックオフ（Exponential Backoff）を用いて待機時間を段階的に伸ばしながら再試行する必要があります：

# robust_client.py
import asyncio
import websockets

async def connect_with_retry(uri):
    retry_delay = 1
    max_delay = 30
    
    while True:
        try:
            async with websockets.connect(uri) as websocket:
                print("WebSocket接続に成功しました！")
                retry_delay = 1  # 安定接続後に待機時間をリセット
                async for msg in websocket:
                    print(f"データ受信: {msg}")
        except (websockets.exceptions.ConnectionClosed, OSError) as e:
            print(f"接続が切断されました ({e})。{retry_delay}秒後に再試行します...")
            await asyncio.sleep(retry_delay)
            retry_delay = min(retry_delay * 2, max_delay)

3. NginxによるリバースプロキシとSSL化（WSS）

Pythonのポートをインターネットへ直接公開するのは避けるべきです。Nginxの背後に配置してSSL終端とリクエストの転送を行います：

location /ws/ {
    proxy_pass http://127.0.0.1:8765;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "Upgrade";
    proxy_set_header Host $host;
    proxy_read_timeout 300s;
}

加えて、ソケットリソースを割り当てる前に不正アクセスを遮断できるよう、最初のハンドシェイク段階でクエリパラメータやヘッダーのJWTトークンを検証することを推奨します。

Share:
Tags: