ZeroMQ via C#: Introduction

• Download binaries - 377.6 KB
• Download source code - 1.9 MB

My Articles about ZeroMQ

• ZeroMQ via C#: Introduction
• ZeroMQ via C#: Multi-part messages, JSON and Synchronized PUB-SUB pattern
• ZeroMQ: Diving Into The Wire

Introduction

ZeroMQ (also spelled ØMQ, 0MQ or ZMQ) is a very lightweight message queuing open source software. It doesn't have a stand-alone server; messages are sent directly from application to application. It is very simple to learn and implement. It is composed of one single library called libzmq.dll written in C++ that can be linked to any application. To use it in the .NET environment, we need a wrapper for this library which is called clrzmq.dll written in C#.

ZeroMQ can be run on Windows, OS X, and Linux. Several languages can be used to implement applications using ZeroMQ including C, C++, C#, Java, Python… This gives the ability to communicate with different applications on different platforms.

The Heart of ZeroMq

The main part of ZeroMQ is the socket. It's not the traditional socket, but it's a socket that provides a layer of abstraction on top of the traditional socket API, which frees us from the complexity and the repeated tasks that we do in our applications. ZeroMQ supports several types of sockets (the type of the socket is defined as an attribute value in the socket itself). The different combinations of socket types in the sending and receiving ends give us different communication patterns, some of which we will explore in this article.

Asynchronous Communication

The communications made by ZeroMQ are done in an asynchronous way. That means our application will not be blocked during setting up or closing the socket connection, re-connection and message delivery. These operations are managed by ZeroMQ itself in background threads and in parallel to the regular processing done by our application. It queues messages (either at sender or receiver side) automatically when needed. It does this intelligently, pushing messages as close as possible to the receiver before queuing them.

Transport Protocols

ZeroMQ supports 4 types of transport protocols. Each transport is defined by an address string which consists of two parts: transport://endpoint. The transport part specifies the underlying transport protocol to use, and the endpoint part is defined according to the used protocol as follows:

1. TCP (tcp://hostname:port): communication over the network
2. INROC (inproc://name): communication within the same process (between threads)
3. IPC (ipc:///tmp/filename): inter-process communication within the same host
4. PGM (pgm://interface;address:port and epgm://interface;address:port): multicast communication over the network

Message Formats

The default message types that we can send or receive are string and array of bytes. ZeroMQ does not impose any format on messages that are sent between sockets. We are free to encode our messages; we can use XML, JSON, MessagePack… In this article, we will use only strings for simplicity.

Binaries

ZeroMq Projects

The libraries that we need are two DLLs.

libzmq.dll

It's a C++ library containing all ZeroMQ stuff. You can get the source code from http://www.zeromq.org/area:download. To build it, follow these steps:

1. You need Microsoft Visual C++ 2008 or later.
2. Unpack the .zip source archive.
3. In Visual C++, open the solution builds\msvc\msvc10.sln.
4. Select 'Release' in the 'Solution Configurations' on the tool bar.
5. Build the solution.
6. ZeroMQ library (libzmq.dll) will be generated in the bin\win32 sub-directory.

clrzmq.dll

It's a .NET wrapper of libzmq.dll library written in C#. You can get the source code from https://github.com/zeromq/clrzmq (get version 3.0.0.0 beta). To build it, follow these steps:

1. Microsoft Visual C# 2010 Express or later
2. Unpack the .zip source archive.
3. In Visual C#, open the solution src\clrzmq.sln.
4. Select 'Release' in the 'Solution Configurations' on the tool bar.
5. Check the 'XML documentation file' check box in the 'Build' tab in ZeroMQ project in order to generate the XML documentation file.
6. Build the solution.
7. clrzmq.dll library will be generated in the src\ZeroMQ\bin\Release sub-directory
8. libzmq.dll (version 3.2.1-beta 2) library will be in the \lib(\x86)|(\x64). This library has been downloaded by the project via nuGet packages.

I have built this solution and got the two DLLs (x86: Clrzmq.dll and libzmq.dll) and used it in this article solution.

Note: You can get the binaries of the beta release of Clrzmq.dll either from NuGet (you must select “Include Prerelease” in the Manage NuGet Packages Window) or from GitHub. In this package, the libzmq.dll (both x86 and x64) are bundled directly in the assembly and are selectively extracted/loaded on application startup (Thanks to John, the maintainer of clrzmq, for his remark).

ZeroMQ Bundle Project

It's the article's solution that I created containing several small console applications allowing us to test different communication patterns in different situations very easily. It contains the two libraries libzmq.dll and clrzmq.dll that I have got from building the clrzmq solution as I mentioned above. Each of these applications has a set of command line parameters (thanks to Giacomo Stelluti Scala for his CommandLine parser open source library). To visualize these parameters, you can type the application name followed by the switch /?. I have also included some batch files for each communication pattern containing the necessary commands for running the pattern. These batch files facilitate running the same pattern at any time rapidly.

Basic Code

In order to use ZeroMQ in Visual C# projects, we must:

1. Add a reference to clrzmq.dll.
2. Add libzmq.dll file to the project (Add existing Item...) (since clrzmq.dll depends on it).
3. Change the libzmq.dll file properties as follows (in order to copy it to output directory at build time):
   • Build Action: Non
   • Copy to Output Directory: Copy if newer
4. Add a using directive for the clrzmq.dll namespace in your code: using ZeroMQ

Now we can write some code to send or receive messages. Let us examine the following code:

using (var context = ZmqContext.Create())
{
    using (var socket = context.CreateSocket(SocketType.REQ))
    {
        socket.Connect("tcp://127.0.0.1:5000");
        socket.Send("My Reply", Encoding.UTF8);
        var replyMsg = socket.Receive(Encoding.UTF8);
    }
}   

First of all, we create a context; from this context, we can create the famous ZeroMQ socket. The type of socket is defined when we create it. With this socket, we can do one of two things:

• Bind to an endpoint and wait for connections from other sockets.
• Connect to an end point.

The selection between bind or connect depends on the communication pattern that we use (explained later in this article).

At last, we can send or receive messages. As we can see, we used very few lines to establish a communication. These steps are used everywhere in the communication patterns.

Communication Patterns

A communication pattern is a set of connected sockets that specify a message flow. We will use some shapes and symbols for illustrating the connections between the sockets. The following diagram shows the basic connection between sockets:

A rectangle is an application that contains one or more sockets. Each socket can bind or connect to an endpoint. A socket that binds to an endpoint is a socket waiting for connections from other sockets.

You will notice that I have added a delay (can be set to any value in milliseconds) before sending or receiving messages. The purpose of this delay is either to:

• Slow sending messages
• Simulate a busy state
• Give some time for socket connections to be completed before sending messages in order not to lose them

Request/Reply Pattern (REQ/REP)

This pattern has the following characteristics:

• The server uses a socket of type REP and the client uses a socket of type REQ.
• The client sends a request and receives a reply while the server receives a request and sends a reply.
• It allows one client to be connected to one or more servers. In this case, the requests are round-robined among all the servers (Reps), one request is sent to one server and the next request is sent to the next sever and so on.
• State based pattern: The client has to receive a reply for its request before sending another one, and the server has to send a reply before receiving another request.

Let us explore this pattern using two cases of client-server connections.

1. One Client - One Server

In this case, we have one client (Req) connected to one server (Rep). The following diagram illustrates it:

The C# code of the server is:

using (var context = ZmqContext.Create())
{
    using (var socket = context.CreateSocket(SocketType.REP))
    {
        foreach (var bindEndPoint in options.bindEndPoints)
            socket.Bind(bindEndPoint);
        while (true)
        {
            Thread.Sleep(options.delay);
            var rcvdMsg = socket.Receive(Encoding.UTF8);
            Console.WriteLine("Received: " + rcvdMsg);
            var replyMsg = options.replyMessage.Replace("#msg#", rcvdMsg);
            Console.WriteLine("Sending : " + replyMsg + Environment.NewLine);                        
            socket.Send(replyMsg, Encoding.UTF8);
        }
    }
}

The C# code of the client is:

using (var context = ZmqContext.Create())
{
    using (var socket = context.CreateSocket(SocketType.REQ))
    {
        foreach (var connectEndpoint in options.connectEndPoints)
            socket.Connect(connectEndpoint);
        long msgCptr = 0;
        int msgIndex = 0;
        while (true)
        {
            if (msgCptr == long.MaxValue)
                msgCptr = 0;
            msgCptr++;
            if (options.maxMessage >= 0)
                if (msgCptr > options.maxMessage)
                    break;
            if (msgIndex == options.alterMessages.Count())
                msgIndex = 0;
            var reqMsg = options.alterMessages[msgIndex++]
                                .Replace("#nb#", msgCptr.ToString("d2"));
            Thread.Sleep(options.delay);
            Console.WriteLine("Sending : " + reqMsg);
            socket.Send(reqMsg, Encoding.UTF8);
            var replyMsg = socket.Receive(Encoding.UTF8);
            Console.WriteLine("Received: " + replyMsg + Environment.NewLine);
        }
    }
}

Double click on ReqRep_Patttern_1.bat file under bin directory. This batch file contains the following commands:

start "Server (Rep)" cmd /T:8E /k Rep.exe -b tcp://127.0.0.1:5000 -r "#msg# - Reply" -d 0

start "Client (Req)" cmd /T:8F /k Req.exe -c tcp://127.0.0.1:5000 -m "Request  #nb#" -x 5 -d 1000  

The first command will start a new colored (/T:fg command) DOS command window and run the application Rep.exe. The Rep application will bind to endpoint tcp://127.0.0.1:5000 and wait for incoming requests. When a request arrives, it will send a reply composed of the incoming request (#msg# macro) concatenated with the word ‘Reply’. The delay before sending the reply is zero milliseconds (-d switch).

The second command will start a new colored (/T:fg command) DOS command window and run the application Req.exe. The Req application will connect to endpoint tcp://127.0.0.1:5000. Then it sends 5 messages (a word ‘Request’ concatenated with the message number (#nb# macro)). It will wait for 1000 milliseconds before sending each request (-d switch).

After running the above commands, we get the following result:

2. One Client – Two Servers

Here, we have one client (Req) connected to two servers (Rep). The following diagram represents this case:

Double click on ReqRep_Patttern_2.bat file under bin directory. This batch file contains the following commands:

start "Server 1 (Rep)" cmd /T:8E /k Rep.exe -b tcp://127.0.0.1:5000 -r "#msg# Reply 1" -d 0

start "Server 2 (Rep)" cmd /T:8E /k Rep.exe -b tcp://127.0.0.1:5001 -r "#msg# Reply 2" -d 0

start "Client (Req)" cmd /T:8F /k Req.exe 
      -c tcp://127.0.0.1:5000;tcp://127.0.0.1:5001 -m "Request  #nb#" -x 5 -d 1000  

The first two commands will run two instances of Rep application; each instance will wait for connections on different port number (5000 and 5001). The last command will run the Req application which will connect to the two running Reps.

After running the above commands, we get the following result:

We notice that the requests are round-robined between the two servers (Reps), one request is sent to one server and the next request is sent to the other sever. It’s the outgoing routing strategy of the REQ sockets.

Publish/Subscribe Pattern (PUB/SUB)

This pattern has the following characteristics:

• The publisher uses socket of type PUB and the subscriber uses socket of type SUB.
• One publisher can have one or more subscribers.
• On subscriber can be connected to one or more publishers.
• Publishers send messages and subscribers receive them.
• Subscriber must subscribe either to all publisher messages by using SubscribeAll method or to a specific message by using Subscribe method and specify (as parameter) the prefix of the message the subscriber is interested in.
• Subscriber can unsubscribe from either all publisher messages by using UnsubscribeAll method or from a specific message by using Unsubscribe method and specifying the message prefix as the method parameter.
• The message filtering happens at:
  • Publisher side (ZeroMQ 3.x for protocols tcp:// and ipc://). The publisher filters messages before sending them to subscribers.
  • Subscriber side (ZeroMQ 3.x for protocol epgm:// and ZeroMQ 2.x). The subscriber drops the unwanted messages received from publishers.
• If a publisher has no connected subscribers, then messages will be dropped.
• A subscriber which is connected to more than one publisher will receive messages evenly (fair-queuing).

Let us explore this pattern using three cases of publisher-subscriber connections.

1. One Publisher - Two Subscribers (all messages subscription)

In this case, we have one publisher (PUB) having two connected subscribers (SUB). The following diagram represents this case:

The C# code of the publisher:

using (var ctx = ZmqContext.Create())
{
    using (var socket = ctx.CreateSocket(SocketType.PUB))
    {                   
        foreach (var endPoint in options.bindEndPoints)
            socket.Bind(endPoint);
                    
        long msgCptr = 0;
        int msgIndex = 0;
        while (true)
        {
            if (msgCptr == long.MaxValue)
                msgCptr = 0;
            msgCptr++;
            if (options.maxMessage >= 0)
                if (msgCptr > options.maxMessage)
                    break;                        
            if (msgIndex == options.altMessages.Count())
                msgIndex = 0;
            var msg = options.altMessages[msgIndex++].Replace("#nb#", msgCptr.ToString("d2"));     
            Thread.Sleep(options.delay);
            Console.WriteLine("Publishing: " + msg);
            socket.Send(msg, Encoding.UTF8);
        }
    }
} 

And the C# code of the subscriber:

using(var ctx = ZmqContext.Create())
{
    using (var socket = ctx.CreateSocket(SocketType.SUB))
    {
        if (options.subscriptionPrefixes.Count() == 0)
            socket.SubscribeAll();
        else
            foreach (var subscriptionPrefix in options.subscriptionPrefixes)
                socket.Subscribe(Encoding.UTF8.GetBytes(subscriptionPrefix));
 
        foreach (var endPoint in options.connectEndPoints)
            socket.Connect(endPoint);
 
        while (true)
        {
            Thread.Sleep(options.delay);
            var msg = socket.Receive(Encoding.UTF8);
            Console.WriteLine("Received: " + msg);
        }
    }
} 

Double click on PubSub_Pattern_1.bat file under bin directory. This batch file contains the following commands:

start "Subscriber 1" cmd /T:8E /k Sub.exe -c tcp://127.0.0.1:5000 -d 0

start "Subscriber 2" cmd /T:8E /k Sub.exe -c tcp://127.0.0.1:5000 -d 0

start "Publisher" cmd /T:8F /k Pub.exe -b tcp://127.0.0.1:5000 
      -m "Orange #nb#";"Apple  #nb#" -x 5 -d 1000 

The first two commands will run two instances of subscriber application (Sub.exe). Each subscriber will connect to endpoint tcp://127.0.0.1:5000 and subscribe to all publisher messages (we did not define a subscription prefix which is the default value). The third command will run the publisher (Pub.exe) which will bind to endpoint tcp://127.0.0.1:5000 and wait for connections from subscribers. Then, it sends 5 messages (alternates between the word ‘Orange’ and ‘Apple’), each of these words is concatenated with the message number (#nb# macro). The delay between messages is 1000 milliseconds (-d switch).

After running the above batch file, we get the following result:

2. One Publisher - Two Subscribers (specific message subscription)

In this case, we have two subscribers (Sub) connected to one publisher (Pub). The first subscriber will subscribe to receive messages that starts with the word “Orange” or “Apple” and the second one will subscribe to receive messages that start with the word “Kiwi”. The following diagram represents this case:

Double click on PubSub_Pattern_2.bat file under bin directory. This batch file contains the following commands:

start "Subscriber 1" cmd /T:8E /k Sub.exe -c tcp://127.0.0.1:5000 -s "Orange";"Apple" -d 0

start "Subscriber 2" cmd /T:8E /k Sub.exe -c tcp://127.0.0.1:5000 -s "Kiwi" -d 0

start "Publisher" cmd /T:8F /k Pub.exe -b tcp://127.0.0.1:5000 
      -m "Orange #nb#";"Apple  #nb#";"Kiwi   #nb#" -x 7 -d 1000 

Note the –s switch in the subscribers’ command line specifies the subscription prefixes.

After running the above batch file, we get the following result:

3. Two Publishers - One Subscriber

In this case, we have one subscriber (Sub) connected to two publishers (Pub). The following diagram represents this case:

Double click on PubSub_Pattern_3.bat file under bin directory. This batch file contains the following commands:

start "Subscriber" cmd /T:8E /k Sub.exe -c tcp://127.0.0.1:5000;tcp://127.0.0.1:5001 -d 0

start "Publisher 1" cmd /T:8F /k Pub.exe -b tcp://127.0.0.1:5000 
      -m "Orange #nb# (Pub 1)";"Apple  #nb# (Pub 1)" -x 5 -d 1000

start "Publisher 2" cmd /T:8F /k Pub.exe -b tcp://127.0.0.1:5001 
      -m "Orange #nb# (Pub 2)";"Apple  #nb# (Pub 2)" -x 5 -d 1000 

Notice that the subscriber is connected to two different endpoints which represent the two publishers. The subscriber is subscribed to all messages.

After running the above batch file, we get the following result:

The subscriber in this example receives messages evenly from among each connection (publisher), one message from one connection and the next one from the next connection and so on, which is the incoming routing strategy of the SUB socket.

Pipeline Pattern (PUSH/PULL)

This pattern is commonly used when there is a need to do parallel data processing. The pipeline pattern scenario is as follows:

1. Normally, we have a task distributor that pushes messages (tasks) to workers in a round-robin fashion (different task for each worker).
2. When the worker receives the message, it will process it and then it will push it to a sort of task collector that receives the messages (tasks).
3. The messages received by the collector are fair-queued among all connected workers.

This pattern has the following characteristics:

• The task distributor uses socket of type PUSH. It binds to its endpoint and waits to receive connections from workers.
• A worker has two sockets, one socket is of type PULL connected to the task distributor socket and the other socket is of type PUSH connected to the collector socket.
• The task collector has a socket of type PULL. It binds to its endpoint and waits to receive connections from workers.

The following diagram represents this pattern:

The C# code of the Task distributor:

using(var ctx = ZmqContext.Create())
{
    using (var socket = ctx.CreateSocket(SocketType.PUSH))
    {
        foreach (var endPoint in options.bindEndPoints)
            socket.Bind(endPoint);
 
        long msgCptr = 0;
        int msgIndex = 0;
        while (true)
        {
            if (msgCptr == long.MaxValue)
                msgCptr = 0;
            msgCptr++;
            if (options.maxMessage >= 0)
                if (msgCptr > options.maxMessage)
                    break;
            if (msgIndex == options.altMessages.Count())
                msgIndex = 0;
            var msg = options.altMessages[msgIndex++].Replace("#nb#", msgCptr.ToString("d2"));
            Thread.Sleep(options.delay);
            Console.WriteLine("Pushing: " + msg);
            socket.Send(msg, Encoding.UTF8);
        }
    }
}

And the C# code of the worker:

using(var ctx = ZmqContext.Create())
{
    using (ZmqSocket receiver = ctx.CreateSocket(SocketType.PULL),
                     sender = ctx.CreateSocket(SocketType.PUSH))
    {
        receiver.Connect(options.pullEndPoint);
        sender.Connect(options.pushEndPoint);
 
        while (true)
        {
            var rcvdMsg = receiver.Receive(Encoding.UTF8);
            Console.WriteLine("Pulled : " + rcvdMsg);
            var sndMsg = options.rcvdMessageTag.Replace("#msg#", rcvdMsg);
            Thread.Sleep(options.delay);
            Console.WriteLine("Pushing: " + sndMsg);
            sender.Send(sndMsg, Encoding.UTF8);
        }
    }
}

And the C# code of the Task collector:

using(var ctx = ZmqContext.Create())
{
    using (var socket = ctx.CreateSocket(SocketType.PULL))
    {
        foreach (var endPoint in options.bindEndPoints)
            socket.Bind(endPoint);
 
        while (true)
        {
            Thread.Sleep(options.delay);
            var msg = socket.Receive(Encoding.UTF8);
            Console.WriteLine("Received: " + msg);
        }
    }
}

Double click on Pipeline_Pattern.bat file under bin directory. This batch file contains the following commands:

start "Task Distributor (Push)" cmd /T:8F /k Push.exe -b tcp://127.0.0.1:5000 
      -m "Orange #nb#";"Apple  #nb#" -x 5 -d 1000

start "Task Collector (Pull)" cmd /T:8E /k Pull.exe -b tcp://127.0.0.1:5001 -d 0

start "Worker 1" cmd /T:1F /k PullPushWorker.exe -l tcp://127.0.0.1:5000 
      -s tcp://127.0.0.1:5001 -t "#msg# (Worker 1)" -d 0

start "Worker 2" cmd /T:1F /k PullPushWorker.exe -l tcp://127.0.0.1:5000 
      -s tcp://127.0.0.1:5001 -t "#msg# (Worker 2)" -d 0 

The first command will run the Task Distributor which will bind to endpoint tcp://127.0.0.1:5000 and wait for connections. The second command will run the Task Collector which will bind to endpoint tcp://127.0.0.1:5001 and wait for connections. The third and forth commands will run two instances of workers. Each worker will connect to the Task Distributor and collector.

After running the above batch file, we get the following result:

Notice that:

• the task distributor has distributed tasks between the connected workers, different task for each worker (round-robin), which is the outgoing routing strategy of the PUSH socket.
• the task collector has received the processed tasks from workers evenly (fair-queue), which is the incoming routing strategy of the PULL socket.

With this pattern, we can simply add other workers without changing any configuration in the task distributor and collector since the workers are connected (not bounded) to their endpoints. We can say that the distributor and the collector are the stable parts of the pattern and the workers are the dynamic parts.

Conclusion

ZeroMQ is a very lightweight open source library. With a few simple lines of code, we can build a communication pattern used on the same or over multiple machines having the same or different platforms. The different parts of a pattern can be implemented by the same or different languages.