How To Work with the ZeroMQ Messaging Library

ZeroMQ is a library used to implement messaging and communication systems between applications and processes - fast and asynchronously.

If you have past experience with other application messaging solutions such as RabbitMQ, it might come a little bit challenging to understand the exact position of ZeroMQ.

When compared to some much larger projects, which offer all necessary parts of enterprise messaging, ZeroMQ remains as just a lightweight and fast tool to craft your own.

Although technically not a framework, given its functionality and the key position it has for the tasks it solves, you can consider ZeroMQ to be the backbone for implementing the actual communication layer of your application.

In this article, we aim to offer you some examples to inspire you with all the things you can do.

Note: We will be working with the Python language and its classic interpreter (Python C interpreter) in our examples. After installing the necessary language bindings, you should be able to simply translate the code and use your favorite instead without any issues. If you would like to learn about installing Python on a CentOS VPS, check out our How to set up Python 2.7 on CentOS 6.4 tutorial.

Create a “server.py” using nano (nano server.py) and paste the below self-explanatory contents.

import zmq

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.REP)
sock.bind("tcp://127.0.0.1:5678")

# Run a simple "Echo" server
while True:
    message = sock.recv()
    sock.send("Echo: " + message)
    print "Echo: " + message

When you are done editing, save and e xit by pressing CTRL+X followed with Y.

Create a “client.py” using nano (nano client.py) and paste the below contents.

import zmq
import sys

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.REQ)
sock.connect("tcp://127.0.0.1:5678")

# Send a "message" using the socket
sock.send(" ".join(sys.argv[1:]))
print sock.recv()

When you are done editing, save and exit by pressing CTRL+X followed with Y.

Note: When working with ZeroMQ library, remember that each thread used to send a message (i.e. .send(..)) expects a .recv(..) to follow. Failing to implement the pair will cause exceptions.

Our server.py is set to work as an “echoing” application. Whatever we choose to send to it, it will send it back (e.g. “Echo: message”).

Run the server using your Python interpreter:

python server.py

On another window, send messages using the client application:

python client.py hello world!
# Echo: hello world!

Note: To shut down the server, you can use the key combination: Ctrl+C

Create a “pub.py” using nano (nano pub.py) and paste the below contents.

import zmq
import time

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.PUB)
sock.bind("tcp://127.0.0.1:5680")

id = 0

while True:
    time.sleep(1)
    id, now = id+1, time.ctime()
    
    # Message [prefix][message]
    message = "1-Update! >> #{id} >> {time}".format(id=id, time=now)
    sock.send(message)
    
    # Message [prefix][message]
    message = "2-Update! >> #{id} >> {time}".format(id=id, time=now) 
    sock.send(message)
    
    id += 1

When you are done editing, save and exit by pressing CTRL+X followed with Y.

Create a “sub.py” using nano (nano sub.py) and paste the below contents.

import zmq

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.SUB)

# Define subscription and messages with prefix to accept.
sock.setsockopt(zmq.SUBSCRIBE, "1")
sock.connect("tcp://127.0.0.1:5680")

while True:
    message= sock.recv()
    print message

When you are done editing, save and exit by pressing CTRL+X followed with Y.

Note: Using the .setsockopt(..) procedure, we are subscribing to receive messages starting with string 1. To receive all, leave it not set (i.e. "").

Our pub.py is set to work as a publisher, sending two different messages - simultaneously - intended for different subscribers.

Run the publisher to send messages:

python pub.py

On another window, see the print outs of subscribed content (i.e. 1):

python sub.py!
# 1-Update! >> 1 >> Wed Dec 25 17:23:56 2013

Note: To shut down the subscriber and the publisher applications, you can use the key combination: Ctrl+C

Very similar in the way it looks to the Publish/Subscribe pattern, the third in line Pipeline pattern comes as a solution to a different kind of problem: distributing messages upon demand.

Use-cases:

Pipelining pattern can be used in cases where are list of queued items need to be routed (i.e. pushed in line) for the one asking for it (i.e. those who pull).

Socket type(s) used:

• zmq.PUSH
• zmq.PULL

Create a “manager.py” using nano (nano manager.py) and paste the below contents.

import zmq
import time

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.PUSH)
sock.bind("tcp://127.0.0.1:5690")

id = 0

while True:
    time.sleep(1)
    id, now = id+1, time.ctime()

    # Message [id] - [message]
    message = "{id} - {time}".format(id=id, time=now)

    sock.send(message)

    print "Sent: {msg}".format(msg=message)

The file manager.py will act as a task allocator.

Create a “worker_1.py” using nano (nano worker_1.py) and paste the below contents.

import zmq

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
sock = context.socket(zmq.PULL)
sock.connect("tcp://127.0.0.1:5690")

while True:
    message = sock.recv()
    print "Received: {msg}".format(msg=message)

The file worker_1.py will act as a task processes (consumer/worker).

Our manager.py is set to have a role of an allocator of tasks (i.e. a manager), **PUSH**ing the items. Likewise, worker_1.py set to work as a worker instance receives these items, when it’s done processing by **PULL**ing down the list.

Run the publisher to send messages:

python manager.py

On another window, see the print outs of subscribed content (i.e. 1):

python worker_1.py!
# 1-Update! >> 1 >> Wed Dec 25 17:23:56 2013

Note: To shut down the subscriber and the publisher applications, you can use the key combination: Ctrl+C

Create a “bind.py” using nano (nano bind.py) and paste the below contents.

import zmq

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
socket = context.socket(zmq.PAIR)
socket.bind("tcp://127.0.0.1:5696")

When you are done editing, save and exit by pressing CTRL+X followed with Y.

Create a “connect.py” using nano (nano connect.py) and paste the below contents.

import zmq

# ZeroMQ Context
context = zmq.Context()

# Define the socket using the "Context"
socket = context.socket(zmq.PAIR)
socket.connect("tcp://127.0.0.1:5696")

When you are done editing, save and exit by pressing CTRL+X followed with Y.

You can use the above example to create any bidirectional uni-connection communication applications.

Note: To shut down either, you can use the key combination: Ctrl+C

<div class=“author”>Submitted by: <a href=“https://twitter.com/ostezer”>O.S. Tezer</a></div>