Posted inPython

ZeroMQ (pyzmq) と Python: ブローカーレスで超高速な通信システムを構築する

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

なぜ ZeroMQ は特別なのか？

RabbitMQ や Kafka のような、メモリを大量に消費するクラスタの構成や維持に疲れているなら、ZeroMQ (ZMQ) がその答えです。従来のメッセージブローカーが中間サーバーを必要とするのに対し、ZMQ は各ノードを独立した実体へと変えます。これは、通常のソケットを「強化版ソケット (sockets on steroids)」にアップグレードするようなものです。

ZMQ は本質的に、非常に軽量なネットワーキングライブラリです。TCP/UDP 経由で生のバイトデータを手動で送信する代わりに、in-process（プロセス内）、inter-process（プロセス間）、TCP、マルチキャストなど、さまざまなプロトコルを介してメッセージをそのまま送信できます。C++ で記述されたコアにより、ZMQ は通常 100 マイクロ秒未満という、集中型ブローカーソリューションを遥かに凌ぐ極めて低いレイテンシを実現します。

「ブローカーレス」という哲学が、最大の利点です。システムの中央サーバーにボトルネックが発生することはありません。拡張が必要な場合は、新しいノードを追加するだけです。pyzmq ライブラリは、この強力な機能を非常に使いやすいインターフェースで Python に提供します。

クイックインストール

pyzmq の導入は非常に簡単です。pip パッケージに必要なバイナリが含まれているため、OS に libzmq を別途インストールする必要はありません。

pip install pyzmq

準備が整っているか、バージョンを素早く確認しましょう：

import zmq
print(f"ZMQ Version: {zmq.zmq_version()}")

実践的な 3 つの古典的通信モデル

ZeroMQ は、単一の接続タイプを強制しません。課題に応じて、適切な「メッセージングパターン」を選択できます。

1. Request-Reply (REQ-REP): リクエスト・レスポンス型

これは HTTP に似た同期的なインタラクションですが、ソケット上で動作します。クライアントがリクエストを送信し、サーバーが応答します。ただし、注意が必要です。ZMQ は状態管理が非常に厳格です。クライアントが応答を待たずに 2 つ続けてメッセージを送信しようとすると、システムは即座にエラーをスローします。

サーバー (server.py):

import zmq
import time

context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://*:5555")

while True:
    # クライアントからのメッセージ受信を待機
    message = socket.recv_string()
    print(f"リクエストを受信: {message}")
    
    # ビジネスロジックの処理をシミュレート
    time.sleep(1)
    socket.send_string(f"処理完了: {message}")

2. Publish-Subscribe (PUB-SUB): データ配信

このモデルは、株価配信システムやログの集中管理に非常に適しています。パブリッシャー（Publisher）がデータをプッシュし、必要なノードがサブスクライブ（Sub）します。重要な注意点として、新しいサブスクライバーが接続する際、TCP ハンドシェイク（Slow Joiner 現象）のために最初の数ミリ秒が失われることがあります。その間に送信されたメッセージは永久に失われます。

パブリッシャー:

import zmq
import time

import random

context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://*:5556")

while True:
    topic = random.choice(["SENSORS", "LOGS"])
    val = random.randint(20, 30)
    # サブスクライバーがデータをフィルタリングできるようにトピックを付与して送信
    socket.send_string(f"{topic} {val}")
    time.sleep(0.1)

3. Push-Pull (Pipeline): 負荷分散

100万枚の画像を処理する必要がある場合は、Push-Pull を使用してください。PUSH ノードがタスクをストリームに投入し、PULL ノード（ワーカー）がそれを受け取ります。ZMQ はラウンドロビン方式で自動的に調整を行うため、追加の負荷分散ロジックを実装しなくても、各ワーカーに均等にタスクが割り振られます。

複雑なデータを抽出するワーカーを作成する際、クラッシュを避けるために正規表現（Regex）のチェックは非常に重要です。Python コードに組み込む前に、Toolcraft の Regex Tester を使用してキャプチャグループのパターンを素早くテストできます。これにより、複雑な文字列のデバッグ時間を大幅に短縮できます。

実務における運用ノウハウ

ZMQ は強力ですが、分散型であるため、どのメッセージが滞留しているかを確認できるダッシュボードはありません。以下の 3 つのルールを適用してください。

  • Monitor Socket の使用: socket.get_monitor_socket() を使用して、EVENT_CONNECTED や EVENT_DISCONNECTED などのイベントをキャッチします。これは、ネットワークが頻繁に変動する Docker/K8s 環境で実行する際に不可欠です。
  • HWM (High Water Mark) の設定: デフォルトでは、ZMQ はメッセージをバッファに保存します。サブスクライバーの処理が遅すぎると、メモリがいっぱいになります。SNDHWM と RCVHWM を設定して待機メッセージ数を制限し、RAM のオーバーフローを防ぎましょう。
  • タイムアウト処理: スクリプトを無限に待機させないでください。zmq.POLLIN を使用するか RCVTIMEO を設定して、相手からの応答がない場合にプログラムが終了またはエラーを報告できるようにします。

接続イベントのキャッチ例:

monitor = socket.get_monitor_socket()
if monitor.poll(timeout=100):
    evt = zmq.utils.monitor.parse_monitor_message(monitor.recv())
    if evt['event'] == zmq.EVENT_CONNECTED:
        print("接続が確立されました！")

結論として、ZeroMQ はあらゆるケースで RabbitMQ を完全に置き換えるものではありません。しかし、純粋な パフォーマンス、柔軟性、そしてインフラの最小化を優先するのであれば、pyzmq は間違いなく最良の選択肢です。

Share:
Tags: