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.
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:
If you start getting errors when you run this, try referring to this article.
References