It’s quite common to do some sort of I/O operation (e.g. REST call) whenever a message is consumed by a RabbitMQ client. This should be done asynchronously, but it’s not as simple as changing the event handling code to async void.

In “The Dangers of async void Event Handlers“, I explained how making an event handler async void will mess up the message order, because the dispatcher loop will not wait for a message to be fully processed before calling the handler on the next one.

While that article provided a workaround that is great to use with older versions of the RabbitMQ Client library, it turns out that there is an AsyncEventingBasicConsumer as from RabbitMQ.Client 5.0.0-pre3 which works great for asynchronous message consumption.

AsyncEventingBasicConsumer Example

First, we need to make sure that the RabbitMQ client library is installed.

Install-Package RabbitMQ.Client

Then, we can set up a publisher and consumer to show how to use the AsyncEventingBasicConsumer. Since this is just a demonstration, we can have both in the same process:

        static void Main(string[] args)
        {
            var factory = new ConnectionFactory() { DispatchConsumersAsync = true };
            const string queueName = "myqueue";

            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                channel.QueueDeclare(queueName, true, false, false, null);

                // consumer

                var consumer = new AsyncEventingBasicConsumer(channel);
                consumer.Received += Consumer_Received;
                channel.BasicConsume(queueName, true, consumer);

                // publisher

                var props = channel.CreateBasicProperties();
                int i = 0;

                while (true)
                {
                    var messageBody = Encoding.UTF8.GetBytes($"Message {++i}");
                    channel.BasicPublish("", queueName, props, messageBody);
                    Thread.Sleep(50);
                }
            }
        }

There is nothing really special about the above code except that we’re using AsyncEventingBasicConsumer instead of EventingBasicConsumer, and that the ConnectionFactory is now being set up with a suspicious-looking DispatchConsumersAsync property set to true. The ConnectionFactory is using defaults, so it will connect to localhost using the guest account.

The message handler is expected to return Task, and this makes it very easy to use proper asynchronous code:

        private static async Task Consumer_Received(object sender, BasicDeliverEventArgs @event)
        {
            var message = Encoding.UTF8.GetString(@event.Body);

            Console.WriteLine($"Begin processing {message}");

            await Task.Delay(250);

            Console.WriteLine($"End processing {message}");
        }

The messages are indeed processed in order:

How to Mess This Up

Remember that DispatchConsumersAsync property? I haven’t really found any documentation explaining what it actually does, but we can venture a guess after a couple of experiments.

First, let’s keep that property, but use a synchronous EventingBasicConsumer instead (which also means changing the event handler to have a void return type). When we run this, we get an error:

It says “In the async mode you have to use an async consumer”. Which I suppose is fair enough.

So now, let’s go back to using an AsyncEventingBasicConsumer, but leave out the DispatchConsumersAsync property:

var factory = new ConnectionFactory();

This time, you’ll see that the the event handler is not firing (nothing is being written to the console). The messages are indeed being published, and the queue is remaining at zero messages, so they are being consumed (you’ll see them accumulate if you disable the consumer).

This is actually quite dangerous, yet there is no error like the one we saw earlier. It means that if a developer forgets to set that DispatchConsumersAsync property, then all messages are lost. It’s also quite strange that the choice of how to dispatch messages to the consumer (i.e. sync or async) is a property of the connection rather than the consumer, although presumably it would be a result of some internal plumbing in the RabbitMQ Client library.

Summary

AsyncEventingBasicConsumer is great for having pure asynchronous RabbitMQ consumers, but don’t forget that DispatchConsumersAsync property.

It’s only available since RabbitMQ.Client 5.0.0-pre3, so if you’re on an older version, use the workaround described in “The Dangers of async void Event Handlers” instead.

When using async/await, you’ll want to use async Task methods most of the time, and use async void methods only for event handlers (see “Async/Await – Best Practices in Asynchronous Programming“, by Stephen Cleary, MSDN Magazine, March 2013).

This conventional wisdom works great if you’re building something like a WPF (GUI) application, and event handlers are invoked occasionally as a result of user interaction (e.g. user presses a button, and an event fires). However, there is another class of event handlers that are invoked as part of a dispatcher loop in a third-party library. async void can be pretty dangerous in these cases.

async void Event Handlers in RabbitMQ

Let’s take the example of RabbitMQ. We’ll set up a basic publisher and consumer. A fundamental property of message queues is that messages are delivered in order, and that’s what we expect to happen.

First, install the RabbitMQ Client library via NuGet:

Install-Package RabbitMQ.Client

Then, we can set up a basic publisher and consumer:

        static void Main(string[] args)
        {
            Console.Title = "async RabbitMQ";

            var factory = new ConnectionFactory();

            using (var connection = factory.CreateConnection())
            using (var channel = connection.CreateModel())
            {
                const string queueName = "testqueue";

                // create queue if not already there

                channel.QueueDeclare(queueName, true, false, false, null);

                // publish

                var props = channel.CreateBasicProperties();
                    
                for (int i = 0; i < 5; i++)
                {
                    var msgBytes = Encoding.UTF8.GetBytes("Message " + i);
                    channel.BasicPublish("", queueName, props, msgBytes);
                }

                // set up consumer

                var consumer = new EventingBasicConsumer(channel);
                consumer.Received += Consumer_Received;
                channel.BasicConsume("testqueue", true, consumer);

                Console.ReadLine();
            }
        }

Our consumer will call the Consumer_Received event handler whenever a message is received. This is the first version of the event handler:

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

            Console.WriteLine("Began handling " + content);

            Thread.Sleep(1000);

            Console.WriteLine("Finished handling " + content);
        }

If we run this now, the messages are processed one at a time and in order just as we expect:

Now, let’s change the event handler to an asynchronous one:

        private static async void Consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            var body = e.Body;
            var content = Encoding.UTF8.GetString(body);

            Console.WriteLine("Began handling " + content);

            await Task.Delay(1000);

            Console.WriteLine("Finished handling " + content);
        }

If we run this now…

…we see that our concurrency and ordering guarantees have just gone out the window.

Understanding async void, Revisited

In my recent article, “In-Depth Async in Akka .NET: Why We Need PipeTo()“, I explained what happens when you call async void methods. Let’s recap that.

Say we have this program. We’re calling an async void method in a loop.

        static void Main(string[] args)
        {
            Console.Title = "async void";

            for (int i = 0; i < 5; i++)
                RunJob("Job " + i);

            Console.ReadLine();
        }

        static async void RunJob(string str)
        {
            Console.WriteLine("Start " + str);

            await Task.Delay(1000);

            Console.WriteLine("End " + str);
        }

When you call an async void method, it’s done in a fire-and-forget manner. The caller has no way of knowing whether or when the operation ended, so it just resumes execution immediately, rather than waiting for the async void method to finish. So you end up with parallel and interleaved execution such as this:

If we change RunJob() to be synchronous instead…

        static void RunJob(string str)
        {
            Console.WriteLine("Start " + str);

            Thread.Sleep(1000);

            Console.WriteLine("End " + str);
        }

…you’ll see that everything happens one at a time and in order:

So you have to be really careful when using async void:

1. There is no way for the caller to await completion of the method.
2. As a result of this, async void calls are fire-and-forget.
3. Thus it follows that async void methods (including event handlers) will execute in parallel if called in a loop.
4. Exceptions can cause the application to crash (see the aforementioned article by Stephen Cleary for more on this).

Fixing async void event handlers

Despite these problems, if you want to await in your event handler, you have to make it async void. To prevent parallel and interleaved execution, you have to lock. However, you can’t await in a lock block, so you need to use a different synchronisation mechanism such as a semaphore.

My own Dandago.Utilities provides a ScopedAsyncLock that allows you to neatly wrap the critical section in a using block:

        private static ScopedAsyncLockFactory factory = new ScopedAsyncLockFactory();

        private static async void Consumer_Received(object sender, BasicDeliverEventArgs e)
        {
            using (var scopedLock = await factory.CreateLockAsync())
            {
                var body = e.Body;
                var content = Encoding.UTF8.GetString(body);

                Console.WriteLine("Began handling " + content);

                await Task.Delay(1000);

                Console.WriteLine("Finished handling " + content);
            }
        }

Like this, messages are consumed one at a time, and in order:

ScopedAsyncLockFactory uses a semaphore underneath, so don’t forget to dispose it!