ZeroMQ を Python から利用する場合のサンプルコードを記載します。

Fixing the World

To fix the world, we needed to do two things. One, to solve the general problem of "how to connect any code to any code, anywhere". Two, to wrap that up in the simplest possible building blocks that people could understand and use easily.
http://zguide.zeromq.org/py:all#Fixing-the-World

• 適宜参照するための公式ドキュメント

インストール例

sudo pip install zmq

サーバー、クライアント Hello World

server.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from time import sleep
from zmq import Context, REP

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

    while True:
        # クライアントからのメッセージ待ち
        message = socket.recv()
        print("Received request: %s" % message)
        sleep(1)
        # 応答 (バイト列)
        socket.send(b"World")

if __name__ == '__main__':
    Main()

client.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from zmq import Context, REQ

def Main():
    context = Context()
    print("Connecting to hello world server...")
    socket = context.socket(REQ)
    socket.connect("tcp://localhost:5555")

    for i in range(10):
        socket.send(b"Hello")
        message = socket.recv()
        print("Received reply %s [ %s ]" % (i, message))

if __name__ == '__main__':
    Main()

TCP 上の zmq のプロトコルで通信する必要があるため、そのままメッセージを送信しても応答はありません。また、サーバが起動していない状態でクライアントを起動しても、エラーとはならずにサーバが起動するまで待機します。

echo 'hello' | nc localhost 5555
↑接続はできても、応答が得られません。

参考: socket を利用する場合

TCP/IP ソケットプログラミングを行う場合は以下のようになります。

server.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from socket import socket, AF_INET, SOCK_STREAM

def Main():
    s = socket(AF_INET, SOCK_STREAM)
    s.bind(('', 5555))
    s.listen(1) # backlog: the maximum length to which the queue of pending connections may grow.
    conn, addr = s.accept() # 1接続と、1backlog → 2コネクションまで確立
    print('Connected by', addr)
    while True:
        data = conn.recv(1024)
        if not data: break
        conn.sendall(data) # echo-back
    conn.close()
    s.close()

if __name__ == '__main__':
    Main()

client.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from socket import socket, AF_INET, SOCK_STREAM

def Main():
    s = socket(AF_INET, SOCK_STREAM)
    s.connect(('localhost', 5555))
    s.sendall(b'Hello world')
    data = s.recv(1024)
    print('Received', repr(data))
    s.close()

if __name__ == '__main__':
    Main()

zmq の場合と異なり、以下のコマンドでもサーバからの応答が得られます。

echo 'hello' | nc localhost 5555

複数の zmq ソケットを監視 (zmq-poll)

複数のファイルディスクリプタからのイベントを待つ poll と似た構造で実装されている、複数の zmq ソケットからのイベント待ちを行う zmq-poll が利用できます。

server_poll.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from time import sleep
from zmq import Context, REP, Poller, POLLIN

def Main():
    context = Context()

    socket1 = context.socket(REP)
    socket1.bind("tcp://*:15555")

    socket2 = context.socket(REP)
    socket2.bind("tcp://*:25555")

    poller = Poller()
    poller.register(socket1, POLLIN)
    poller.register(socket2, POLLIN)

    while True:
        try:
            socks = dict(poller.poll()) # クライアントからのメッセージ待ち
        except KeyboardInterrupt:
            break

        for socket in socks:
            message = socket.recv()
            print("Received request: %s" % message)
            sleep(1)
            socket.send(b"World")

if __name__ == '__main__':
    Main()

client1.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from zmq import Context, REQ

def Main():
    context = Context()
    print("Connecting to hello world server...")
    socket = context.socket(REQ)
    socket.connect("tcp://localhost:15555")

    for i in range(10):
        socket.send(b"Hello-1")
        message = socket.recv()
        print("Received reply %s [ %s ]" % (i, message))

if __name__ == '__main__':
    Main()

client2.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from zmq import Context, REQ

def Main():
    context = Context()
    print("Connecting to hello world server...")
    socket = context.socket(REQ)
    socket.connect("tcp://localhost:25555")

    for i in range(10):
        socket.send(b"Hello-2")
        message = socket.recv()
        print("Received reply %s [ %s ]" % (i, message))

if __name__ == '__main__':
    Main()

実行例

サーバ

$ python server_poll.py
Received request: Hello-1
Received request: Hello-1
Received request: Hello-1
Received request: Hello-2
Received request: Hello-2
Received request: Hello-2
Received request: Hello-1
Received request: Hello-2
Received request: Hello-1
Received request: Hello-2
Received request: Hello-1
Received request: Hello-2
Received request: Hello-1
Received request: Hello-2
...

クライアント1

$ python client1.py
Connecting to hello world server...
Received reply 0 [ World ]
Received reply 1 [ World ]
Received reply 2 [ World ]
Received reply 3 [ World ]
Received reply 4 [ World ]
...

クライアント2

$ python client2.py
Connecting to hello world server...
Received reply 0 [ World ]
Received reply 1 [ World ]
Received reply 2 [ World ]
Received reply 3 [ World ]
...

zmq poll で生の TCP ソケットを監視

zmq ソケットだけでなく、TCP ソケットのイベントも監視することができます。

zmq_poll() shall examine either the ØMQ socket referenced by socket or the standard socket specified by the file descriptor fd
http://api.zeromq.org/3-2:zmq-poll

server_poll_2.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from time import sleep
from zmq import Context, REP, Poller, POLLIN

from socket import socket as tpc_socket, AF_INET, SOCK_STREAM

def Main():
    context = Context()

    socket1 = context.socket(REP)
    socket1.bind("tcp://*:15555")

    socket2 = context.socket(REP)
    socket2.bind("tcp://*:25555")

    tcp_socket = tpc_socket(AF_INET, SOCK_STREAM)
    tcp_socket.bind(('', 35555))
    tcp_socket.listen(1)

    poller = Poller()
    poller.register(socket1, POLLIN)
    poller.register(socket2, POLLIN)
    poller.register(tcp_socket, POLLIN)

    while True:
        try:
            socks = dict(poller.poll()) # クライアントからのメッセージ待ち
        except KeyboardInterrupt:
            break

        for socket in socks:
            if socket == tcp_socket.fileno():
                conn, addr = tcp_socket.accept()
                print('Established TCP connection with', addr)
                while True:
                    data = conn.recv(1024)
                    if not data: break
                    conn.sendall(data)
                conn.close()
            else:
                message = socket.recv()
                print("Received request: %s" % message)
                sleep(1)
                socket.send(b"World")

if __name__ == '__main__':
    Main()

client3.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

from socket import socket, AF_INET, SOCK_STREAM

def Main():
    s = socket(AF_INET, SOCK_STREAM)
    s.connect(('localhost', 35555))
    s.sendall(b'Hello world')
    data = s.recv(1024)
    print('Received', repr(data))
    s.close()

if __name__ == '__main__':
    Main()

Pub-Sub パターン

クライアント側は unicode でフィルター文字列を指定できます。また Pub-Sub パターンでは接続前に Publish されたメッセージを取得しなくてもよい Subscriber を想定して実装します。

server.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import zmq

def Main():
    context = zmq.Context()
    socket = context.socket(zmq.PUB)
    socket.bind('tcp://*:5556')
    while True:
        socket.send_string('aaa bbb ccc')
        socket.send_string('xxx yyy zzz')

if __name__ == '__main__':
    Main()

client.py

#!/usr/bin/python
# -*- coding: utf-8 -*-

import zmq

def Main():
    context = zmq.Context()
    socket = context.socket(zmq.SUB)
    socket.connect('tcp://localhost:5556')

    filter = u'xxx'
    socket.setsockopt_string(zmq.SUBSCRIBE, filter)

    for i in range(5):
        string = socket.recv_string()
        print(string)

if __name__ == '__main__':
    Main()

実行例

python client.py
xxx yyy zzz
xxx yyy zzz
xxx yyy zzz
xxx yyy zzz
xxx yyy zzz