Nicholas Piël

ZeroMQ is a messaging library, which allows you to design a complex communication system without much effort. It has been wrestling with how to effectively describe itself in the recent years. In the beginning it was introduced as ‘messaging middleware’ later they moved to ‘TCP on steroids’ and right now it is a ‘new layer on the networking stack’.

I had some trouble understanding ZeroMQ at first and really had to reset my brain. First of all, it is not a complete messaging system such as RabbitMQ or ActiveMQ. I know the guys of Linden Research compared them, but it is apples and oranges. A full flexed messaging system gives you an out of the box experience.  Unwrap it, configure it, start it up and you’re good to go once you have figured out all its complexities.

ZeroMQ is not such a system at all; it is a simple messaging library to be used programmatically. It basically gives you a pimped socket interface allowing you to quickly build your own messaging system.

Float like a butterfly, sting like a bee

But why use ZeroMQ and not just use the low level Berkeley socket interface or a high level messaging system? I think the answer is balance. You probably want the flexibility and performance of the low level while still having the ease of implementation of the high level. However, maintaining raw sockets is difficult and cumbersome when you want to implement a scalable system. A high level system often works perfect if you use it for the situation it was designed for, but it can be difficult to change core elements of the system and its ease of use often comes with a cost in performance. This isn’t a problem that is limited to messaging systems only. We can see the previous dilemma also in web frameworks; it could very well be that this is exactly the reason why ‘Micro Frameworks’ gain in popularity.

I believe that ZeroMQ perfectly fits this gap between the high and the low level, so what are its features?

Performance

ZeroMQ is blazing fast.  It is orders of magnitude faster than most AMQP messaging systems and it can obtain this high performance because of the following techniques:

• It does not have the overhead of an over-engineered protocol such as AMQP
• It can make use of efficient transports such as reliable Multicast or the Y-suite IPC transport
• It makes use of intelligent message batching. This allows 0MQ to efficiently utilize a TCP/IP connection by minimizing not only protocol overhead but also system calls.

Simplicity

The API is deceptively simple, and it makes sending messages really simple compared with a raw socket implementation where you have to continuously ‘feed’ the socket buffer. In ZeroMQ you can just fire off an async send call, it will queue the message in a separate thread and do all the work for you.  Because of this async nature, your application does not have to waste time waiting until the message has been flushed.  The async nature of 0MQ makes it a perfect companion for an event-based framework.

ZeroMQ’s simple wire protocol fits perfectly in the current time setting where we have lots of different transport protocols. With AMQP it always felt a bit weird to use an extra protocol layer on top. 0MQ gives you complete freedom on how you encode your message, as it will just interpret it as a blob. So you can send simple JSON messages, go the binary route with for example BSON, Protocol Buffers or Thrift and all this without feeling guilty.

Scalability

While ZeroMQ sockets look low level they provide lots of features. A single ZeroMQ socket can for example connect to multiple end points and automatically load balance messages over them. Or it can work as some sort of Fan-In, collecting messages from multiple sources through a single socket.

ZeroMQ  follows a brokerless design so that there is no single point of failure. Combine this with its simplicity and performance and you get something that you can use to make your application distributed.

Implementing a messaging layer with ZeroMQ

In the next section I will show how to design and implement a messaging layer with ZeroMQ.  For the code example I will use Brian Granger’s PyZMQ, which is the excellent Python binding to ZeroMQ.

Implementing a ZeroMQ messaging layer is a three-step approach:

1. Choose a transport
2. Set up the infrastructure
3. Select a messaging pattern

Choosing a transport

The first step is to choosing a transport. ZeroMQ provides 4 different transports:

1. INPROC an In-Process communication model
2. IPC an Inter-Process communication model
3. MULTICAST multicast via PGM, possibly encapsulated in UDP
4. TCP a network based transport

The TCP transport is often the best choice, it is very performant and robust. However, when there is no need to cross the machine border it can be interesting to look at the IPC or INPROC protocol to lower the latency even more. The MULTICAST transport can be interesting in special cases. But personally, I am a bit careful with applying multicast, as it is difficult to understand how it will behave when scaling up. Think of issues such as figuring out how many multicast groups you can create with this or that hardware and how much stress it is going to put on the different switches in your network. If you want to be sure that your code runs cross platforms it is probably best to go with TCP as the other transports are not guaranteed to be available on the different platforms.

Setting up the infrastructure

When you have decided upon your transport you will have to think about how the different components are connected to each other. It is simply answering the question: “Who connects to whom?”  You probably want the most stable part of the network to BIND on a specific port and have the more dynamic parts CONNECT to that. In the image below we have depicted how a server binds to a certain port and how a client connects to it.

It is possible that both ends of the networks are relatively dynamic so that it is difficult to have a single stable connection point. If this is the case, you could make use of the forwarding devices that ZeroMQ provides. These devices can bind to 2 different ports and forward messages from one end to the other. By doing so, the forwarding device can become the stable point in your network where each component can connect to. ZeroMQ provides three kinds of devices:

1. QUEUE, a forwarder for the request/response messaging pattern
2. FORWARDER, a forwarder for the publish/subscribe messaging pattern
3. STREAMER, a forwarder for the pipelined messaging pattern

In the image below we can see such a device being used, in this situation both the client and the server initialize a connection to the forwarder, which binds to two different ports. Using such a device will remove the need of extra application logic, as you will not need to maintain a list of connected peers.

Selecting a message pattern

The previous steps build the infrastructure but did not specify the message flow. The next step is to think carefully about the message pattern each component should follow. The patterns that 0MQ supports are:

1. REQUEST/REPLY, bidirectional, load balanced and state based
2. PUBLISH/SUBSCRIBE, publish to multiple recipients at once
3. UPSTREAM / DOWNSTREAM, distribute data to nodes arranged in a pipeline
4. PAIR, communication exclusively between peers

I will explain them a bit more below.

Request Reply

The request reply paradigm is very common and can be found in most type of servers. For example: HTTP, POP or IMAP. This pattern has a certain state associated with it as a request has to be followed by a reply. The client uses a socket of type REQ as it will initiate the request by performing a .send() on the socket. The server uses a socket of type REP, and it will start by performing a .recv() to read the incoming request, after which it can send its reply.

ZeroMQ greatly simplifies this pattern by allowing you to have a single socket connect to multiple end points. ZeroMQ will automatically balance requests over the different peers.

The Python code below will create an echo server that listens on port 5000 with a REP socket. It will then loop an alternation of performing .recv() for incoming requests and then .send() a reply to them.

import zmq
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.bind("tcp://127.0.0.1:5000")

while True:
    msg = socket.recv()
    print "Got", msg
    socket.send(msg)

When you have multiple clients connected to this server the ZMQ socket will fair queue between all incoming requests. Now, if you want your client to be able to connect to multiple servers as well, you can take the above code, change port 5000 to 6000 and use it to run an extra server. The following client code will then be able to use both of the servers:

import zmq
context = zmq.Context()
socket = context.socket(zmq.REQ)
socket.connect("tcp://127.0.0.1:5000")
socket.connect("tcp://127.0.0.1:6000")

for i in range(10):
    msg = "msg %s" % i
    socket.send(msg)
    print "Sending", msg
    msg_in = socket.recv()

The above sends 10 requests in total but since we are connected to 2 different servers, each server only has to handle 5 requests. Isn’t that great? With only a few lines of code we were able to create a distributed client/server model.

Now, if we want to add an extra server to handle our requests we will have to adjust our code. This can be cumbersome as we need to do this for all our clients to let them know it can now balance the requests over an extra server.

This is exactly where the ZeroMQ devices fit in. Instead of having the clients connect directly to multiple servers it can connect to a single forwarding device. The forwarding device will then reroute all messages to the connected servers.

Example client output:

Sending msg 0
Sending msg 1
Sending msg 2
Sending msg 3
Sending msg 4
Sending msg 5
Sending msg 6
Sending msg 7
Sending msg 8
Sending msg 9

Example output server 1 at port 5000:

Got msg 0
Got msg 2
Got msg 4
Got msg 6
Got msg 8

Example output server 2 at port 6000:

Got msg 1
Got msg 3
Got msg 5
Got msg 7
Got msg 9

Publish Subscribe

The Pub/Sub paradigm has gained lots of interest the last few years. You can think of things such as message pushing, XMPP or webhooks. In a pub/sub pattern the components are loosely coupled. This will greatly help you to scale out as there is no need to worry about the subscribers. However, this loose coupling can also lead to unexpected behavior when not fully understood. A nice metaphor for the Pub/Sub paradigm is thinking of it is a radio station. When you publish messages you send something over a certain frequency, only listeners that have subscribed to that frequency will receive the signal. But also, just as with a radio, if you tuned in to the station after the broadcast you will miss the show.

It is good to stress that the various message patterns have no coupling with the infrastructure. It is thus possible to bind to a port and publish to the peers that connect to it. But it is also possible to do it the other way around, connect to multiple peers and broadcast to them. The first example resembles the radio metaphor (everybody can tune in), while the second one more resembles yelling at your peers through a megaphone (a selected group). In both situations your peers can decide not to listen to your messages by not subscribing to them.

The following code shows how you could create a broadcasting server for live soccer events:

import zmq
from random import choice
context = zmq.Context()
socket = context.socket(zmq.PUB)
socket.bind("tcp://127.0.0.1:5000")

countries = ['netherlands','brazil','germany','portugal']
events = ['yellow card', 'red card', 'goal', 'corner', 'foul']

while True:
    msg = choice( countries ) +" "+ choice( events )
    print "->",msg
    socket.send( msg )

The server will generate an unlimited amount of events for the different countries and pushes them over a socket of type PUB. Below you can find some example output:

-> portugal corner
-> portugal yellow card
-> portugal goal
-> netherlands yellow card
-> germany yellow card
-> brazil yellow card
-> portugal goal
-> germany corner
…

Now if we are only interested in events concerning The Netherlands and Germany we can create a client that subscribes to those specific messages:

import zmq

context = zmq.Context()
socket = context.socket(zmq.SUB)
socket.connect("tcp://127.0.0.1:5000")
socket.setsockopt(zmq.SUBSCRIBE, "netherlands")
socket.setsockopt(zmq.SUBSCRIBE, "germany")

while True:
    print  socket.recv()

The client will create a SUB socket, connect to our broadcast server at port 5000 and subscribe to messages starting with ‘netherlands’ or ‘germany’. The output will look something like this:

netherlands red card
netherlands goal
netherlands red card
germany foul
netherlands yellow card
germany foul
netherlands goal
netherlands corner
germany foul
netherlands corner
…

Pipelining

The pipeline pattern looks remarkably similar to the Rep/Req pattern, the difference is that instead of requiring a reply being sent to the requester the reply can be pushed down the pipe. This is a paradigm commonly seen when there is a need to process data  in parallel. For example, lets say we have some sort of system that does face recognition. We have a job server that pushes the images to one of the workers, which will then process it, once finished it will then push it down the stream again towards some sort of collector.

In the design at the left we can see that a worker will receive its message from an UPSTREAM socket and once they are processed sends them DOWNSTREAM. It routes messages from two different socket types.

The jobserver can just keep pushing tasks DOWNSTREAM through a single socket but with multiple endpoints. ZeroMQ and recently also PyZMQ can send the messages in a zero-copy manner. This is great if you need to push large messages around and you don’t want to waste IO cycles.

Paired sockets

Paired sockets are very similar to regular sockets as the communication is bidirectional, there is no specific state stored within the socket and there can only be one connected peer. Most real life problems can be captured in one of the previously explained patterns and I want to recommend that you look at them first before applying this one as it will simplify your problem.

The figure at the left depicts the infrastructure of a paired socket, the server listens on a certain port and a client connects to it. The red lines indicate the flow of messages, in this pattern both endpoints use a socket of type PAIR and as you can see the messages can flow bidirectional.

The following code shows how to implement such a thing. We will bind to a port on one end:

import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.bind("tcp://127.0.0.1:5555")

And on the other end where we will connect to it.

import zmq
context = zmq.Context()
socket = context.socket(zmq.PAIR)
socket.connect("tcp://127.0.0.1:5555")

ZeroMQ and the future

In this post I have given a short introduction to ZeroMQ, I hope that at this point you will now share my ideas about what a great little library it is. But while the library may feel small it has a grand vision of being the new messaging layer. And really, it is not that weird when you come to think of it. Scalability issues are mostly just communication and portability issues, ZeroMQ can solve these problems for you.

Lets say you want to create some new sort of database because Redis, Cassandra, TokyoTyrant, Postgres, MongoDB, DabbleDB, CouchDB, HBase, etc. just don’t serve your needs that well. You create an amazing in memory tree representation for your data and have a blazing fast indexer. Now all you need is some sort of messaging layer such that different clients can talk to your server. Preferably implemented in different programming language and with clustering capabilities. You could of course create such a messaging framework all by yourself, but that is a lot of hard work.

A simple solution is to just implement your database as a ZeroMQ server and pick a message protocol (fe JSON). As you have seen by now, implementing such functionality with ZeroMQ is really easy and on top of this you will get almost instant scalability because of the way ZeroMQ can route messages. It will also make it incredibly easy to implement different clients that will communicate with your server. Basically all you need to do is pick one of the 15 available language bindings, use the same message protocol and you’re done. Currently the following languages have a ZeroMQ binding: Ada, C, C++, Common Lisp, Erlang, Go, Haskell, Java, Lua, .NET, OOC, Perl, PHP, Python and Ruby.

ZeroMQ could very well be the new way in how we connect our components. A good example of someone who understands the possibilities of ZeroMQ is Zed Shaw as can be seen with his recent project Mongrel2. You can use Mongrel2 to bridge the gap between a regular HTTP client and a ZeroMQ component. If you don’t immediately see how awesome this is you probably have never worked with websockets, comet or flash based sockets. Another way to look at the great possibilities of such an implementation is to think of Facebook’s BigPipe where each Pagelet can transparantly be generated by a different component connected with 0MQ.