ZeroMQ

ちょっとZeroMQを調べてみようと。

ZeroMQとは、組み込みで動くメッセージングミドルウェア…というか、ライブラリのようですね。

ZeroMQを使うことで、様々な通信パターンを実現できるとか。

ØMQ(ZeroMQ, 0MQ, zmq などとも呼ばれます) は組み込みネットワークライブラリと言うことも出来ますが、並行フレームワークの様にも機能します。
ØMQ はプロセス内通信、プロセス間通信、TCP やマルチキャストの様な幅広い通信手段を用いてアトミックにメッセージを転送する通信ソケットを提供します。
ソケットをファンアウト、Pub-Sub、タスク分散、リクエスト・応答の様なパターンで N 対 N で接続できます。非同期 I/O モデルにより、マルチコア環境でスケーラブルな非同期メッセージ処理を行うアプリケーションを構築可能で、製品クラスタを構成する上で十分高速です。

ØMQガイドブック(日本語版)

ZeroMQ覚書

Raspberry Pi3 と ZeroMQ でフォグ・コンピューティングの雰囲気を体験（前編）

REQ-REPパターン

サーバーとクライアントが1対1でやり取りするパターンです。リクエストを送って、レスポンスを受け取る構成ですね。

今回は、こちらを使って簡単なechoプログラムを書いてみます。

環境

環境は、こちら。

$ lsb_release -a
No LSB modules are available.
Distributor ID:	Ubuntu
Description:	Ubuntu 18.04.2 LTS
Release:	18.04
Codename:	bionic


$ python3 -V
Python 3.6.7

言語は、Pythonを利用します。

ライブラリのインストール

Python向けのZeroMQのライブラリは、こちらを使用します。

PyZMQ: Python bindings for ØMQ

$ pip3 install pyzmq

今回利用するライブラリのバージョン。

Successfully installed pyzmq-18.0.1

サンプルプログラム

それでは、サンプルプログラムを書いてみます。

REQ-REPなので、メッセージを受け取ってレスポンスを返すサーバーを書いてみます。

server.py

import zmq

def start_server():
    context = zmq.Context()
    socket = context.socket(zmq.REP)
    socket.bind("tcp://*:5556")

    print("Server startup.")

    while True:
        message = socket.recv_string()
        print("Received message = %s" % message)
        socket.send_string("Reply: %s" % message)

    socket.close()
    context.destroy()

if __name__ == "__main__":
    start_server()

最初にContextを作成して、socketをsocket_type（今回はREP）を指定して作成します。

zmq.Context.socket

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

サーバーとしてリッスンするポートは、5556としました。

あとは、メッセージを待ち、受け取ったらReply:を付けてクライアントに返します。

    while True:
        message = socket.recv_string()
        print("Received message = %s" % message)
        socket.send_string("Reply: %s" % message)

終了する時は、socketのクローズとコンテキストの破棄。

    socket.close()
    context.destroy()

続いて、クライアント側。

client.py

import sys
import zmq

def start_client():
    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:5556")

    while True:
        print("Enter message:")
        message = sys.stdin.readline()
        socket.send_string(message)

        recv_message = socket.recv_string()
        print("Receive message = %s" % recv_message)

    socket.close()
    context.destroy()

if __name__ == "__main__":
    start_client()

クライアントの場合は、socket_typeがREQとなり、socket.connectを使って接続します。

    context = zmq.Context()
    socket = context.socket(zmq.REQ)
    socket.connect("tcp://localhost:5556")

メッセージは、標準入力から読み込んで送信するようにしました。

    while True:
        print("Enter message:")
        message = sys.stdin.readline()
        socket.send_string(message)

        recv_message = socket.recv_string()
        print("Receive message = %s" % recv_message)

確認

サーバーを起動。

$ python3 server.py 
Server startup.

クライアントを起動。

$ python3 client.py 
Enter message:

適当にメッセージを送ってみます。

Enter message:
こんにちは
Receive message = Reply: こんにちは

Enter message:
Hello World
Receive message = Reply: Hello World

Enter message:
こんにちは、世界
Receive message = Reply: こんにちは、世界

サーバー側からメッセージが戻ってきました。

サーバー側のコンソールには、こんな表示が出ます。

Received message = こんにちは

Received message = Hello World

Received message = こんにちは、世界

うまくいったようです。

ところで、この状態でもうひとつクライアントを起動してみます。

$ python3 client.py

メッセージを送ります。

Enter message:
foo
Receive message = Reply: foo

Enter message:
bar
Receive message = Reply: bar

Enter message:
hoge
Receive message = Reply: hoge

サーバー側には送ったメッセージが表示されます。

Received message = foo

Received message = bar

Received message = hoge

ですが、もともと接続していたクライアントには変化がありません。

Enter message:
こんにちは
Receive message = Reply: こんにちは

Enter message:
Hello World
Receive message = Reply: Hello World

Enter message:
こんにちは、世界
Receive message = Reply: こんにちは、世界

Enter message:

ちゃんと1対1で動いているということなんですね。