How to utilize prefetch count for consumer with .NET RabbitMQ client

Question

Its states the following in the RabbitMQ documentatation

"As a rule of thumb, sharing Channel instances between threads is something to be avoided. Applications should prefer using a Channel per thread instead of sharing the same Channel across multiple threads."

Currently we are looking at the prefetch count where it has been recommended that if you have a small number of consumers and autoack=false, then we should consume many messages at once. However we find that the prefetch has not effect if the consumer sends manual acknowledgements back using a single thread of execution. However if we wrap the consumer processing in a Task, we find that the prefetch count does matter and substantially improves consumer performance.

See the following example where we are wrapping the consumption of the message by the consumer in a Task object:

class Program
{
    public static void Main()
    {
        var factory = new ConnectionFactory()
        {
            HostName = "172.20.20.13",
            UserName = "billy",
            Password = "guest",
            Port = 5671,
            VirtualHost = "/",
            Ssl = new SslOption
            {
                Enabled = true,
                ServerName = "rabbit.blah.com",
                Version = System.Security.Authentication.SslProtocols.Tls12
            }
        };
        var connection = factory.CreateConnection();
        var channel = connection.CreateModel();
        channel.BasicQos(0, 100, false);
        channel.ExchangeDeclare(exchange: "logs", type: "fanout");
        var queueName = channel.QueueDeclare().QueueName;
        Console.WriteLine(" [*] Waiting for logs.");

        var consumer = new EventingBasicConsumer(channel);
        consumer.Received += (model, ea) =>
        {
            var _result = new Task(() => {
                var body = ea.Body;
                var message = Encoding.UTF8.GetString(body);
                System.Threading.Thread.Sleep(80);

                channel.BasicAck(ea.DeliveryTag, false);
            });
            _result.Start();
        };
        channel.BasicConsume(queue: "test.queue.1", autoAck: false, consumer: consumer);

        Console.WriteLine(" Press [enter] to exit.");
        Console.ReadLine();
    }
}

The question I have is how do people implement consumers with the .NET rabbitmq client that avail of prefetch counts?, Do you have to manual ack using a Task of some sort?, is it safe

Personally, I hate RabbitMQ API, you find much more bugs along the way (for example even if you say in options that you need to consume one by one message it still consumes multiple messages! damn!). Better to use some message bus API like MassTransit. It hides this hilarious Channel manipulations, Queue management and connection corruptions which can lead to deadlocks. It also has support to mock rabbitmq for testing. — eocron, Feb 08 '19 at 08:27
More here - https://stackoverflow.com/questions/12296787/what-does-masstransit-add-to-rabbitmq — eocron, Feb 08 '19 at 08:33
@eocron - the RabbitMQ .NET client is [open source](https://github.com/rabbitmq/rabbitmq-dotnet-client), and we (the RabbitMQ team) regularly receive pull requests to fix bugs and improve the code. Rather than making claims on stack overflow, why not contribute your apparent expertise to the project? — Luke Bakken, Feb 08 '19 at 14:41
@LukeBakken Is the only way to take advantage of the prefetch count performance benefits by using Tasks to send acks back? — bstack, Feb 11 '19 at 11:45

score 5 · Accepted Answer · answered Feb 11 '19 at 15:21

The documentation to which you refer is for the Java client. You should be referring to this document instead.

You're using the latest version of the .NET client (5.1), so doing your work in the Received event handler won't block other threads that deal with TCP data and it won't block heartbeats - both of which are good.

First of all, calling channel.BasicQos(0, 1, false) means your consumer will only receive one ready message at a time from RabbitMQ, and until BasicAck is called another message won't be delivered. So, there's really no reason to do your work in another thread since you're not going to get another message anyway.

If you increase the prefetch value (through experimentation and running benchmarks), you will have to do your work in a background thread if your work takes more than a few milliseconds to run.

When you do your work in the Received event callback, it's going to block the thread that is used to make that callback since the callback is not executed on its own thread. So, you can either ensure your work is very short, or do the work in another thread.

I just spent some time reviewing the .NET client code and I'm pretty sure the IModel instance is not thread safe. If you increase prefetch, you will have the opportunity to acknowledge several messages at the same time, so I recommend implementing a solution that uses that and also ensures that BasicAck is called on the same thread on which the connection is created.

_{NOTE: the RabbitMQ team monitors the rabbitmq-users mailing list and only sometimes answers questions on StackOverflow.}

Thanks for the response and apologies, I have updated my example to have a prefetch count equal to 100. (this was a typo) — bstack, Feb 12 '19 at 08:33
Does this mean that you are recommending that we use the same thread of execution to perform BasicAck, but set multiple to true? — bstack, Feb 12 '19 at 08:35
@LukeBakken "I recommend implementing ... ensures that BasicAck is called on the same thread on which the connection is created." Did you mean to say on the same thread _that created the channel_, or really the thread that created the connection? — Boris B., Jan 27 '22 at 21:31

score 1 · Answer 2 · answered Feb 08 '19 at 17:09

Source: https://www.rabbitmq.com/api-guide.html

When manual acknowledgements are used, it is important to consider what thread does the acknowledgement. If it's different from the thread that received the delivery (e.g. Consumer#handleDelivery delegated delivery handling to a different thread), acknowledging with the multiple parameter set to true is unsafe and will result in double-acknowledgements, and therefore a channel-level protocol exception that closes the channel. Acknowledging a single message at a time can be safe.

channel.basicAck(tag, false) is thread safe

but consumerChannel.basicAck(tag, true) is not.

Also some good points mentioned at RabbitMQ and channels Java thread safety

That is the Java guide, and the person asking this question is using .NET. There are differences between the libraries. — Luke Bakken, Feb 11 '19 at 14:33

How to utilize prefetch count for consumer with .NET RabbitMQ client

2 Answers2