Zmq

Notes

Zmq (abbreviation for Zero MQ) is messaging queue software.

Messaging queue software follows the pub/sub (publisher/subscriber) architecture. This involves creating asychronous messaging pipelines. Agents that are creating events can publish to a pipeline, while agents that are processing events can subscribe to a pipeline.

Documentation: https://pyzmq.readthedocs.io/

Pair Model: Simple Client and Server

The following two files, pairclient.py and pairserver.py, illustrate the simplest possile pair model for zmq.

pairclient.py:

import zmq
import random
import sys
import time

"""
ZMQ Pair Client

This generates test messages every half second,
faster than the server will print/process the
messages. This just illustrates that ZMQ will
store messages in a queue as they come in, and
will not throw messages away.
"""

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://localhost:%s" % port)

i = 0
while True:
    i += 1
    socket.send_string("ping %d"%(i))
    time.sleep(0.5)

pairserver.py:

import zmq
import random
import sys
import time

"""
ZMQ Pair Server

This accepts messages from a ZMQ message queue
and prints them at a rate of two per five seconds.
This is slower than the client generates messages.
Thils illustrates the queue behavior of ZMQ.
"""

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://*:%s" % port)

while True:
    msg = socket.recv()
    print(msg)

    msg = socket.recv()
    print(msg)

    time.sleep(5)

Pair Model: Json Client and Server

The following example expands on the prior example, and shows how to build a ZMQ client/server pair that send and receive JSON data:

jsonclient.py:

import json
import zmq
import random
import sys
import time

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://localhost:%s" % port)

for i in range(50):
    ## Send as string
    #d = dict(apples=random.randint(1,1000), oranges=random.randint(1,1000), bananas=random.randint(1,1000))
    #print("sending data %s"%(json.dumps(d)))
    #socket.send_string("%s"%json.dumps(d))
    #time.sleep(0.1)

    # Send as dict
    d = dict(apples=random.randint(1,1000), oranges=random.randint(1,1000), bananas=random.randint(1,1000))
    print("sending data %s"%(d))
    socket.send_json(d)
    time.sleep(0.1)

jsonserver.py:

import json
import zmq
import random
import sys
import time

port = "5556"
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://*:%s" % port)

while True:
    ## Receive as string
    #msg = socket.recv()
    #d = json.loads(msg)
    #print(d)
    #time.sleep(1)

    # Receive as json
    d = socket.recv_json()
    print(type(d))
    print(d)
    time.sleep(1)