Category Archives: RabbitMQ

Message Persistence in RabbitMQ and BenchMarkDotNet

(Note: if you want to follow the code on this, you might find it easier if you start from the project that I create here.)

A queue in a message broker can be persistent, which means that, should you have a power failure (or just shut down the server), when it comes back, the queue is still there.

So, we can create a durable (persistent) queue, like this:

var result = channel.QueueDeclare("NewQueue", true, false, false, args);

The second parameter indicates that the queue is durable. Let’s send it some messages:


static void Main(string[] args)
{            
    for (int i = 1; i <= 100; i++)
    {
        string msg = $"test{i}";
 
        SendNewMessage(msg);
    } 
    
}
private static void SendNewMessage(string message)
{
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
        Dictionary<string, object> args = 
            DeadLetterHelper.CreateDeadLetterQueue(channel,
            "dl.exchange", "dead-letter", "DeadLetterQueue");
 
        var result = channel.QueueDeclare("NewQueue", true, false, false, args);
        Console.WriteLine(result);
 
        channel.BasicPublish("", "NewQueue", null, Encoding.UTF8.GetBytes(message));                
 
    }
}

Now we have 100 messages:

persist1

Let’s simulate a server reboot:

parsist2

Following the reboot, it’s gone:

persist3

Admittedly, that doesn’t sound very durable!

Why?

The reason for this, is that the durability of the queue doesn’t affect the durability of the message. At least, if the queue is durable, it doesn’t make the message so.

How can it be made persistent?

Let’s change our send code a little:


private static void SendNewMessage(string message)
{
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
        Dictionary<string, object> args = 
            DeadLetterHelper.CreateDeadLetterQueue(channel,
            "dl.exchange", "dead-letter", "DeadLetterQueue");
 
        var result = channel.QueueDeclare("NewQueue", true, false, false, args);
        Console.WriteLine(result);
 
        IBasicProperties prop = channel.CreateBasicProperties();
        prop.Persistent = true;
 
        channel.BasicPublish("", "NewQueue", prop, Encoding.UTF8.GetBytes(message));                
 
    }
}

The only difference is the addition of the IBasicProperties parameter. The Persistent flag is set. Now we’ll re run the same test; here’s the messages:

persist4

And after restarting the service:

persist5

As you can see, the messages are still there, and you can see the time difference where they’re been restored to the queue after a failure.

Speed – Introducing BenchmarkDotNet

I suppose the main question here is what price do you pay for durability. This gives me a chance to play with a new tool that I heard about a little while ago: BenchmarkDotNet.

It’s quite straightforward to use, just add the NuGet package:

Install-Package BenchmarkDotNet

There’s a bit of refactoring; I effectively ripped out the send and called it from a separate class:


class Program
{        
    static void Main(string[] args)
    {
        BenchmarkRunner.Run<SpeedTest>();
    }
}
 
public class SpeedTest
{
    [Benchmark]
    public void SendNewMessagePersist()
    {
        MessageHelper helper = new MessageHelper();
        helper.SendStringMessage("Test", "NewQueue", true);
    }
 
    [Benchmark]
    public void SendNewMessageNonPersist()
    {
        MessageHelper helper = new MessageHelper();
        helper.SendStringMessage("Test", "NewQueue", false);
    }
 
 
}

I then ran this:

persist6

And it produced this:

persist7

So, it is a bit slower to persist the message. I’m not sure how helpful this information is: I probably could have guessed that persisting the message would have been slower beforehand. Having said that, I am quite impressed with BenchMarkDotNet!

Acknowledging a Message Using RabbitMQ

Conceptually, message queues work in a similar fashion: you send a message to an exchange, the exchange allows people to read the message, and you have functionality that ensures the original message arrives with the destined recipient. Acknowledgement of a message is basically a way to ensure that delivery. Having said that, the two main message brokers that I’ve been investigating deal with this in a slightly different manner (albeit, the same things happen in the end).

If you want to follow through, it might be an idea to use the code from my first post as a starting point.

Let’s change the send code first to send a few messages:

static void Main(string[] args)
{            
    for (int i = 1; i <= 100; i++)
    {
        string msg = $"test{i}";
 
        SendNewMessage(msg);        
    } 
            
}

Now we have 100 messages.

rabbitack1

Next, we’ll look at the receiving code:

public void ReceiveNextMessage()
{
    var result = _channel.QueueDeclare("NewQueue", true, false, false, null);
    Console.WriteLine(result);
 
    EventingBasicConsumer consumer = new EventingBasicConsumer(_channel);
    consumer.Received += Consumer_Received;
 
    _channel.BasicConsume("NewQueue", false, consumer);
 
}

BasicConsume() has a parameter called “noAck”. It took me a while to work this out, but noAck means that it doesn’t expect an acknowledgement; that is, it will automatically acknowledge receipt. So, noAck = True mean automatically acknowledge, and noAck = False means manually acknowledge. That not entirely uncomplicated.

The received event looks a bit like this:

private void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    var body = e.Body;
    var message = Encoding.UTF8.GetString(body);
    //if (message.Contains("3")) 
    //   throw new Exception("Error here !");
 
    //_channel.BasicAck(e.DeliveryTag, false);
 
    Console.WriteLine(message);
}

I’ve left the error and the commented out BasicAck in on purpose. If you run this, unlike with ActiveMQ, where you will get a message at a time, you will get all messages in the queue (because it’s event based). Add in the BasicAck() to acknowledge the queue and you’re good to go.

If you add in the error at this stage, you can see that, in exactly the same way as ActiveMQ, it will only acknowledge the correct message. What you will also see here is we have the poison message scenario that I discussed in this post on ActiveMQ.

IMHO, this is where RabbitMQ beats ActiveMQ hands down. The following code is the simplest version of dealing with a poison message:

private void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    try
    {
        var body = e.Body;
        var message = Encoding.UTF8.GetString(body);
 
        if (message.Contains("3"))
            throw new Exception("Error here !");
 
        _channel.BasicAck(e.DeliveryTag, false);
 
        Console.WriteLine(message);
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex);
        _channel.BasicNack(e.DeliveryTag, false, true);
    }
}

So, we have a problem, and we issue a Nack. What the Nack does is allow you to re-queue the message. The code above allows the queue to process, but just moves all the bad ones to the back. The obvious problem here is that it the problem isn’t transient, we’ll keep coming back to them. It does, however, get around the problem – the queue is no longer blocked.

The solution, as it was with ActiveMQ, is to put them in a “dead letter queue”; however, unlike ActiveMQ, this is remarkably easy. Firstly, let’s refactor our queue creation a little:

public void ReceiveNextMessage()
{
    Dictionary<string, object> args = DeadLetterHelper.CreateDeadLetterQueue(_channel);
 
    // How declare the queue and pass in the dead letter exchange
    var result = _channel.QueueDeclare("NewQueue", true, false, false, args);
    Console.WriteLine(result);
 
    EventingBasicConsumer consumer = new EventingBasicConsumer(_channel);
    consumer.Received += Consumer_Received;
 
    _channel.BasicConsume("NewQueue", false, consumer);
 
}

You’ll notice that we go to a new helper method called: “CreateDeadLetterQueue()” and return a dictionary; which is, in turn, passed through to our new queue. The CreateDeadLetterQueue() function looks like this:

public static Dictionary<string, object> CreateDeadLetterQueue(IModel channel,
    string deadLetterExchange, string deadLetterRoutingKey, string deadLetterQueue)
{
    // Declare dead letter exchange                     
    channel.ExchangeDeclare(deadLetterExchange, "direct");
    Dictionary<string, object> args = new Dictionary<string, object>()
    {
        { "x-dead-letter-exchange", deadLetterExchange },
        { "x-dead-letter-routing-key", deadLetterRoutingKey }
    };
 
    // Bind the exchange to a queue
    channel.QueueDeclare(deadLetterQueue, true, false);
    channel.QueueBind(queue: deadLetterQueue,
                    exchange: deadLetterExchange,
                    routingKey: deadLetterRoutingKey);
    return args;
}

There’s effectively two steps. Firstly we need a dead letter exchange, and this needs a routing key (in this case, “dead-letter”). Next, we declare the DeadLetterQueue with the same routing key. Finally, return the argument list, which allows the linking of the main queue to the dead letter queue.

Now we are going to change the receive code so that it doesn’t re-queue:

private void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    try
    {
        var body = e.Body;
        var message = Encoding.UTF8.GetString(body);
 
        if (message.Contains("3"))
            throw new Exception("Error here !");
 
        _channel.BasicAck(e.DeliveryTag, false);
 
        Console.WriteLine(message);
    }
    catch (Exception ex)
    {
        Console.WriteLine(ex);
        _channel.BasicNack(e.DeliveryTag, false, false);
    }
}

And running it results in a dead letter queue full of all our dodgy data:

rabbitack2

If you start getting errors when you run this, try referring to this article.

References

RabbitMQ documentation on the subject

RabbitMQ Change Queue

Whilst trying to add a dead letter exchange to an existing queue, I got the following error:

Exception thrown: ‘RabbitMQ.Client.Exceptions.OperationInterruptedException’ in RabbitMQ.Client.dll

Additional information: The AMQP operation was interrupted: AMQP close-reason, initiated by Peer, code=406, text=”PRECONDITION_FAILED – inequivalent arg ‘x-dead-letter-exchange’ for queue ‘NewQueue’ in vhost ‘/’: received the value ‘dl.exchange’ of type ‘longstr’ but current is none”, classId=50, methodId=10, cause=

There is a clue is in the error message:

…received the value ‘dl.exchange’ of type ‘longstr’ but current is none

So, the problem is that the exchange is changing.

One solution is to delete the queue:

rabbitmqchange1

This didn’t feel right, so I asked, and it turns out you can’t do this in code.

Also, remember that both the publisher and subscriber need to change the code, otherwise, the publisher will re-declare the queue without the exchange. Not that this happened to me – I realised that straight away, and didn’t spend an hour trying to work out why even deleting the queue didn’t work.

A C# Programmer’s Guide to Installing, Running and Messaging with RabbitMQ

Following this year’s trip to DDD North I got speaking to someone about Rabbit MQ. I’d previously done some research on Active MQ, however, given that there are, realistically, only two options in this market, I thought I should have a look at the competition.

Install from here.

You also need to install Erlang.

After the install… nothing happens. No matter how long you wait, assuming it’s just your slow laptop (don’t want to talk about that anymore). Navigate to:

rabbitmq1

C:\Program Files\RabbitMQ Server\rabbitmq_server-3.6.0\sbin

In a DOS prompt, type:

rabbitmq-plugins enable rabbitmq_management

rabbitmq2

Rabbit will now run as a service, which also feels much cleaner that ActiveMQs console app style interface.

Now, log-in to http://localhost:15672/, and log-in as guest/guest:

rabbitmq3

Code

Just like with ActiveMQ, there is a NuGet package that you can use for interacting with the exchange:

https://www.nuget.org/packages/RabbitMQ.Client. Your project needs .Net 4.5.1 or higher; otherwise, you’ll see this:

rabbitmq4

If it’s not, upgrade your project to 4.5.1:

rabbitmq5

Okay, so now we’re up and running, here’s the code for the send:

static void Main(string[] args)
{            
    while (true)
    {
        string msg = Console.ReadLine();
        if (string.IsNullOrWhiteSpace(msg))
        {
            break;
        }
 
        SendNewMessage(msg);
    } 
    
}
private static void SendNewMessage(string message)
{
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
        var result = channel.QueueDeclare("NewQueue", true, false, false, null);
        Console.WriteLine(result);
 
        channel.BasicPublish("", "NewQueue", null, Encoding.UTF8.GetBytes(message));                
 
    }
}

Just like with ActiveMQ, you can now just look at the queue:

rabbitmq6

A quick caveat here:


var result = channel.QueueDeclare("NewQueue");

Doesn’t work because exclusive defaults to ‘true’, meaning that only your session can see it.

Receiving a Message


static void Main(string[] args)
{
    while (true)
    {
        ReceiveNextMessage();
 
        string exit = Console.ReadLine();
        if (!string.IsNullOrWhiteSpace(exit)) break;
    }
 
}
 
private static void ReceiveNextMessage()
{
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
        var result = channel.QueueDeclare("NewQueue", true, false, false, null);
        Console.WriteLine(result);
 
        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
        consumer.Received += Consumer_Received;
 
        channel.BasicConsume("NewQueue", true, consumer);
 
        System.Threading.Thread.Sleep(1000);
 
    }
}
 
private static void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    var body = e.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine(message);
}

rabbitmq7

Why the Thread.Sleep? The reason is that, because the message consumer is event based, you need to stop tidying up the objects until you’re done. Obviously, sleeping for 1 second has its issues. I suspect this could be better achieved using a TaskCompletionSource, of even just a Console.ReadLine(), but for the purposes of this post, it suffices.

Topics

The concept of topics seems to be the same as ActiveMQ; however, the implementation is different, and there were a few dead-ends to go down first.

The following code, for example:


var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare("NewTopic", "testTopic");
    
    channel.BasicPublish("NewTopic", "test", null, Encoding.UTF8.GetBytes(message));
 
}

Gives the error: COMMAND_INVALID – unknown exchange type ‘testTopic’

rabbitmq8

RabbitMQ seems to have some magic strings that determine the message type. This works:

 
private static void SendNewTopic(string message)
{
var factory = new ConnectionFactory() { HostName = "localhost" };
using (var connection = factory.CreateConnection())
using (var channel = connection.CreateModel())
{
    channel.ExchangeDeclare("NewTopic", "topic"); // topic is a magic string
    
    channel.BasicPublish("NewTopic", "test", null, Encoding.UTF8.GetBytes(message));
 
}

The “test” is a routing code. These seem to be quite an involved system; however, it basically tells the exchange where the message is going.

rabbitmq9

Receive


private static void ReceiveNextTopic()
{
    var factory = new ConnectionFactory() { HostName = "localhost" };
    using (var connection = factory.CreateConnection())
    using (var channel = connection.CreateModel())
    {
        channel.ExchangeDeclare("NewTopic", "topic");
        var queueName = channel.QueueDeclare().QueueName;
 
        channel.QueueBind(queue: queueName,
                        exchange: "NewTopic",
                        routingKey: "test");
 
        EventingBasicConsumer consumer = new EventingBasicConsumer(channel);
        consumer.Received += Consumer_Received;
 
        channel.BasicConsume(queueName, true, consumer);
 
        Console.ReadLine();
    }
}

private static void Consumer_Received(object sender, BasicDeliverEventArgs e)
{
    var body = e.Body;
    var message = Encoding.UTF8.GetString(body);
    Console.WriteLine(message);
}

Notice the routing key; this (broadly speaking) needs to match the sender’s routing key. The purpose seems to be to allow a complex mapping between publisher and subscriber.

Conclusion

In comparison to ActiveMQ, RabbitMQ seems to be a more actively maintained code base; a quick comparison of the GitHub repo of each reveals more activity in Rabbit. Rabbit also seems a little cleaner in the way it’s run and installed; however, it also seems to have more moving parts.

There are a number of companies that offer support for ActiveMQ commercially; however, the company that wrote RabbitMQ offer commercial support, should you need it.

Can you run Active and RabbitMQ together?

No. No, you can’t. They both seem to want to use the same resources, and so if you want to use them both on the same machine, you can, but only one at a time.

Acknowledgements

I used the following websites extensively in this article:

http://arcware.net/installing-rabbitmq-on-windows/

https://www.rabbitmq.com/tutorials/tutorial-one-dotnet.html

Asynchronous Behaviour in Applications using Message Queues

This was one of the features that interested me most in message queues. Basically, you have an application, and you want to communicate to it; that is, Joe Bloggs the user is sat there, tapping away at his keyboard, and I want to send him a message that interrupts him. There are dozens of use cases for this: the user has entered an order and there’s a problem with it, the user’s account has been locked and needs to be logged out, we want to alert them that there’s a data change so that they can refresh their data.

The relevant part of message queuing here is a topic; which allows me to send an alert to one of more listeners.

To this end, I’ve extended my helper class to include a TopicHelper:

public class TopicHelper : IDisposable
{
    IReceiveTopicBehaviour _receiveBehaviour = null;
    ConnectionFactory _factory = new ConnectionFactory() { HostName = "localhost" };
    IConnection _connection = null;
    IModel _channel = null;
    EventingBasicConsumer _consumer = null;
 
    private void SetupChannel()
    {
        if (_connection == null)
        {
            _connection = _factory.CreateConnection();
            _channel = _connection.CreateModel();
        }
    }
 
    private void SetupConsumer()
    {
        if (_channel == null) throw new Exception("Channel is not set-up");
        if (_connection == null) throw new Exception("Connection is not set-up");
 
        if (_consumer == null)
            _consumer = new EventingBasicConsumer(_channel);
    }
 
    public void ReceiveTopic(string topicName, string key, IReceiveTopicBehaviour behaviour)
    {
        _receiveBehaviour = behaviour;
 
        SetupChannel();
        SetupConsumer();
 
        _channel.ExchangeDeclare(topicName, "topic");
        var queue = _channel.QueueDeclare();
        var queueName = queue.QueueName;
 
        _channel.QueueBind(queue: queueName,
        exchange: topicName,
        routingKey: key);
 
        _consumer.Received += Consumer_Received;
 
        _channel.BasicConsume(queueName, true, _consumer);
 
    }
 
    private void Consumer_Received(object sender, BasicDeliverEventArgs e)
    {
        var body = e.Body;
        var message = Encoding.UTF8.GetString(body);
        Console.WriteLine(message);
 
        _receiveBehaviour.OnReceive(message);
    }
 
    public void SendTopic(string topicName, string key, string data)
    {
        SetupChannel();
 
        _channel.ExchangeDeclare(topicName, "topic");
        _channel.BasicPublish(topicName, key, null, Encoding.UTF8.GetBytes(data));            
 
    }
 
    public void Dispose()
    {            
        _connection.Dispose();
        _channel.Dispose();            
    }
}

There’s a lot of code here, but basically there’s only two methods of note: SendTopic() and ReceiveTopic(). I’ve also made it a disposable class. The next thing we want is a listener; for this, I’ve used a WPF app, but any application should be able to do this:

rabbitalert1

The call to set-up the alert is this:

public partial class MainWindow : Window
{
    private TopicHelper _topicHelper = new TopicHelper();
    public MainWindow()
    {
        InitializeComponent();
 
        Task.Run(() =>
        {
            ReceiveTopic recTopic = new ReceiveTopic();
            _topicHelper.ReceiveTopic("Alerts", "thisApp", recTopic);
        });
    }
}

I’ve used the code behind because I’m just proving a point. Obviously, in real life, this would be some abstraction in the business layer. The main thing to note is the ReceiveTopic class that is instantiated and passed through; here’s its implementation:


public class ReceiveTopic : IReceiveTopicBehaviour
{
    public void OnReceive(string message)
    {
        System.Windows.MessageBox.Show(message);
    }
}

This, effectively, allows me to provide custom functionality on the receive. If you find that you’re using this to pass in the same one-liner, you could adapt this to simply pass in an action.

The final piece of the puzzle is the send alert; in my case this is a console app:

rabbitalert2

Here’s the code for the main function:


static void Main(string[] args)
{
    Console.WriteLine("Alert: ");
    string alertText = Console.ReadLine();
 
    using (TopicHelper helper = new TopicHelper())
    {
        helper.SendTopic("Alerts", "thisApp", alertText);
    }
 
    Console.WriteLine("Finished");
    Console.ReadLine();
}

Does it work?

Yes, of course.

rabbitalert3