Python websockets ライブラリ徹底解説：リアルタイム通信を簡単に実装！ 🚀

WebSocketは、単一のTCP接続上でクライアントとサーバー間の全二重（双方向）通信を可能にする通信プロトコルです。2011年にIETFによってRFC 6455として標準化されました。

従来のHTTP通信は、基本的にクライアントがリクエストを送信し、サーバーがそれに応答するという「リクエスト・レスポンス」モデルです。リアルタイムな通信を実現するために、HTTPポーリング（定期的にクライアントから問い合わせる）やロングポーリング（サーバー側で更新があるまで接続を保持する）といった手法が使われてきましたが、これらは以下のような課題がありました。

• 遅延: ポーリングでは、次の問い合わせまでの間にタイムラグが発生します。
• オーバーヘッド: 毎回HTTPリクエストを送信するため、ヘッダーなどの不要なデータ通信量が多くなります。
• サーバー負荷: 特にロングポーリングでは、多数の接続を長時間維持する必要があり、サーバーリソースを消費します。

WebSocketはこれらの課題を解決します。

WebSocketの仕組み

1. ハンドシェイク: 最初に、クライアントは通常のHTTPリクエストに似た形式でサーバーに接続要求を送ります。このリクエストには Upgrade: websocket や Connection: Upgrade といった特殊なヘッダーが含まれており、WebSocketプロトコルへの切り替えを要求します。
2. プロトコル切り替え: サーバーがWebSocketをサポートしていれば、ステータスコード 101 Switching Protocols を含むHTTPレスポンスを返します。これにより、既存のTCP接続がHTTPからWebSocketプロトコルにアップグレードされます。
3. 双方向通信: ハンドシェイクが成功すると、接続は確立されたまま維持されます。クライアントとサーバーは、どちらからでも自由に、低オーバーヘッドな「フレーム」と呼ばれる単位でデータを送受信できるようになります。TCP接続が切断されるまで、この状態が続きます。

WebSocketのメリット ✨

• 低遅延: 接続が確立されていれば、いつでもデータを送受信できるため、リアルタイム性が高いです。
• 効率的: HTTPヘッダーのようなオーバーヘッドが少ないため、通信量が削減されます。
• 双方向性: クライアントとサーバーの両方から能動的にデータを送信できます。
• ステートフル: 接続が維持されるため、接続ごとの状態管理が容易になります。

websockets ライブラリのインストールは pip を使って簡単に行えます。ターミナルまたはコマンドプロンプトで以下のコマンドを実行してください。

pip install websockets

現在の最新バージョン（2025年4月時点）は 15.0.1 で、Python 3.9 以上が必要です。

コード解説:

• import asyncio と from websockets.server import serve で必要なモジュールをインポートします。
• echo(websocket) 関数は、クライアント接続ごとに呼び出されるハンドラーです。websocket オブジェクトを通じて通信を行います。
• async for message in websocket: は、クライアントからメッセージが送られてくるのを待ち受け、受信するたびにループが回ります。
• await websocket.send(message) で、受信したメッセージをそのままクライアントに送り返します。
• try...except ブロックで、接続の切断（正常またはエラー）をハンドリングしています。
• main() 関数内で websockets.serve(echo, "localhost", 8765) を使ってサーバーを起動します。第一引数にハンドラー関数、第二引数にホスト名、第三引数にポート番号を指定します。
• async with serve(...) により、サーバーの起動と停止が適切に管理されます。
• await asyncio.Future() は、プログラムが終了せずにサーバーが動作し続けるようにするための一般的な方法です。
• asyncio.run(main()) で非同期イベントループを開始し、main コルーチンを実行します。
• 注意: サーバーを外部ネットワークからアクセス可能にする場合は、ホスト名に 'localhost' ではなく '0.0.0.0' を指定します。

コード解説:

• from websockets.client import connect でクライアント接続用の関数をインポートします。
• uri = "ws://localhost:8765" で接続先のサーバーURIを指定します。
• async with connect(uri) as websocket: でサーバーへの接続を試みます。connect() も非同期コンテキストマネージャーとして機能し、接続の確立と切断を管理します。
• 接続が成功すると、await websocket.send(...) でメッセージを送信し、await websocket.recv() でサーバーからの応答を待ち受けます。
• try...except ConnectionRefusedError で、サーバーが起動していないなどの理由で接続できなかった場合のエラーを処理します。

• 非同期デザイン: asyncio をベースにしているため、多数の接続を効率的に扱えます。
• 接続管理: 接続の確立、維持、切断を自動的に処理します。ハンドシェイクやPing/Pong（接続維持確認）も内部で管理されます。
• メッセージ形式: テキストメッセージとバイナリメッセージの両方をサポートします。
• Graceful Shutdown: 接続を正常に閉じるための仕組みを提供します。
• 圧縮 (Compression): permessage-deflate 拡張 (RFC 7692) をサポートし、通信データを圧縮して帯域幅を節約できます。
• サブプロトコル (Subprotocols): クライアントとサーバー間でアプリケーションレベルのプロトコルをネゴシエートできます。
• カスタムヘッダー: ハンドシェイク時にカスタムHTTPヘッダーを追加できます。
• セキュリティ (WSS): wss:// URI を使用することで、SSL/TLSによる暗号化通信をサポートします。
• エラーハンドリング: 接続エラーやプロトコル違反など、様々な例外クラスを提供し、詳細なエラーハンドリングを可能にします。

6.1. 複数クライアントの管理とブロードキャスト

サーバーは通常、複数のクライアントからの接続を同時に処理する必要があります。また、特定のクライアントから受信したメッセージを、接続中の他の全クライアントに送信（ブロードキャスト）したい場合もよくあります。

import asyncio
import websockets
import json
import logging

# ロギングの設定
logging.basicConfig(level=logging.INFO)

# 接続中のクライアントを管理するセット
connected_clients = set()

# ブロードキャスト用の非同期関数
async def broadcast(message):
    # disconnected_clients = set() # ブロードキャスト中に切断されたクライアントを追跡
    # 送信処理を並行して行う
    tasks = [client.send(message) for client in connected_clients]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    # 送信に失敗した（切断された可能性のある）クライアントを特定
    # for i, result in enumerate(results):
    #     if isinstance(result, Exception):
    #         disconnected_clients.add(list(connected_clients)[i]) # エラーが発生したクライアントを追加

    # 切断されたクライアントをセットから削除（より堅牢な方法が必要な場合あり）
    # if disconnected_clients:
    #     logging.info(f"切断されたクライアントを削除: {len(disconnected_clients)} 件")
    #     connected_clients.difference_update(disconnected_clients)

# クライアント接続ハンドラー
async def handler(websocket):
    # 新しいクライアントをセットに追加
    connected_clients.add(websocket)
    logging.info(f"クライアント接続: {websocket.remote_address}, 現在の接続数: {len(connected_clients)}")

    try:
        # クライアントからのメッセージを待機
        async for message in websocket:
            logging.info(f"受信: {message} from {websocket.remote_address}")
            # ここでは受信メッセージをそのまま全クライアントにブロードキャスト
            # 実際にはメッセージ内容に応じて処理を行う
            await broadcast(f"{websocket.remote_address[0]}:{websocket.remote_address[1]} より: {message}")
    except websockets.ConnectionClosedOK:
        logging.info(f"クライアント切断 (正常): {websocket.remote_address}")
    except websockets.ConnectionClosedError as e:
        logging.warning(f"クライアント切断 (エラー): {websocket.remote_address}, code={e.code}, reason={e.reason}")
    finally:
        # クライアントが切断されたらセットから削除
        connected_clients.remove(websocket)
        logging.info(f"接続終了: {websocket.remote_address}, 残りの接続数: {len(connected_clients)}")
        # 他のクライアントに退出を通知
        await broadcast(f"ユーザー {websocket.remote_address[0]}:{websocket.remote_address[1]} が退出しました。")


async def main():
    async with websockets.server.serve(handler, "localhost", 8765):
        logging.info("チャットサーバー起動 (ws://localhost:8765)")
        await asyncio.Future()  # run forever

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

この例では、connected_clients というセットを使って接続中のクライアント (websocket オブジェクト) を管理します。新しい接続があれば追加し、切断されれば削除します。broadcast 関数は、このセット内の全クライアントに対してメッセージを送信します。asyncio.gather を使うことで、複数の送信処理を効率的に並行実行できます。

6.2. 定期的なデータ送信

サーバーからクライアントへ定期的にデータをプッシュしたい場合があります（例: 株価情報、センサーデータなど）。asyncio.create_task と asyncio.sleep を組み合わせることで実現できます。

import asyncio
import websockets
import datetime
import random
import json

connected_clients = set()

# 定期的にデータを送信するタスク
async def send_periodic_data(websocket):
    while websocket in connected_clients: # 接続が維持されている間ループ
        try:
            now = datetime.datetime.utcnow().isoformat() + "Z"
            data = {
                "timestamp": now,
                "value": random.random() * 100
            }
            await websocket.send(json.dumps(data))
            await asyncio.sleep(5) # 5秒待機
        except websockets.ConnectionClosed:
            # 送信中に接続が切れた場合
            break
        except Exception as e:
            logging.error(f"定期送信エラー: {e}")
            break

async def handler(websocket):
    connected_clients.add(websocket)
    logging.info(f"クライアント接続: {websocket.remote_address}, 現在の接続数: {len(connected_clients)}")

    # 定期送信タスクを開始
    send_task = asyncio.create_task(send_periodic_data(websocket))

    try:
        # クライアントからのメッセージも受信可能 (この例では受信メッセージは無視)
        async for message in websocket:
            logging.info(f"受信 (無視): {message} from {websocket.remote_address}")
            pass # 受信メッセージに対する処理はここでは行わない
    except websockets.ConnectionClosedOK:
        logging.info(f"クライアント切断 (正常): {websocket.remote_address}")
    except websockets.ConnectionClosedError as e:
        logging.warning(f"クライアント切断 (エラー): {websocket.remote_address}, code={e.code}, reason={e.reason}")
    finally:
        # 定期送信タスクをキャンセル
        send_task.cancel()
        connected_clients.remove(websocket)
        logging.info(f"接続終了: {websocket.remote_address}, 残りの接続数: {len(connected_clients)}")

async def main():
    # ... (サーバー起動部分は前述の例と同じ) ...
    async with websockets.server.serve(handler, "localhost", 8765):
        logging.info("定期データ送信サーバー起動 (ws://localhost:8765)")
        await asyncio.Future()

if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO)
    asyncio.run(main())

この例では、クライアントが接続すると send_periodic_data コルーチンが別のタスクとして開始され、5秒ごとにランダムなデータをJSON形式で送信します。クライアントが切断されると、このタスクもキャンセルされます。

6.3. エラーハンドリング

ネットワークの問題、プロトコル違反、アプリケーション固有のエラーなど、様々なエラーが発生する可能性があります。websockets は詳細な例外クラスを提供しており、これらを適切に捕捉して処理することが重要です。

• ConnectionClosedOK: 正常に接続が閉じられた場合に発生。
• ConnectionClosedError: エラーにより接続が閉じられた場合に発生。コード (例: 1002 Protocol Error) と理由が含まれます。
• ConnectionClosed: 上記2つの基底クラス。
• ProtocolError: WebSocketプロトコルに違反があった場合に発生。
• その他、タイムアウト関連の例外など。

堅牢なアプリケーションを構築するためには、これらの例外を適切に処理し、ログを記録し、必要に応じてリトライ処理やユーザーへの通知を行う必要があります。

6.4. デプロイメント

開発したWebSocketアプリケーションを本番環境で動作させる場合、いくつかの点を考慮する必要があります。

• リバースプロキシ: NginxやApacheなどのリバースプロキシサーバーを前面に配置し、WebSocket接続をプロキシすることが一般的です。これにより、HTTPS (WSS) の終端、負荷分散、静的ファイルの配信などを効率的に行えます。
• プロセス管理: Supervisorやsystemdなどのプロセス管理ツールを使用して、WebSocketサーバープロセスをデーモン化し、監視、自動再起動などを行うことが推奨されます。
• スケーリング: 高負荷に対応するためには、サーバープロセスを複数起動し、リバースプロキシで負荷分散を行うか、Centrifugoのような専用のリアルタイムメッセージングサーバーとの連携を検討します。
• セキュリティ: WSS (TLS/SSL) を使用して通信を暗号化することが不可欠です。また、認証・認可の仕組みを適切に実装し、不正なアクセスを防ぐ必要があります。