How does the "ConsumerDispatchConcurrency" work? (v7.2.0)

[Oliver-Piorun]

Hey!

• We want to specify the ConsumerDispatchConcurrency on a channel-level (in contrast to specifying it on a connection-level)
• How to get a single, dedicated dispatch thread per channel?
  • Setting the value to 1 should be equal to the fallback (connection-level ConsumerDispatchConcurrency)
  • This would mean, that no dedicated dispatch thread is being used but still the single one that is shared across all channels
  • Do you have to set this value to 2 to get a dedicated dispatch thread at all?
  • Setting it to 2 would mean that you get 2 dedicated dispatch threads for that channel and still keep the shared one for the other channels, I guess? Or do you get a dedicated one and the shared dispatch thread will also be used?

Thanks in advance! :>

[lukebakken]

How to get a single, dedicated dispatch thread per channel?

TL;DR - this is what this client library implements by default.

I wanted to use your question to test the genie (AI), and it provided a very good overview of how consumer dispatch works. Let me know if you have additional questions about this feature.

---

Based on my examination of the RabbitMQ .NET client v7.2.0 codebase, here's how ConsumerDispatchConcurrency works:

Architecture Overview

The ConsumerDispatchConcurrency setting controls how many concurrent worker tasks process consumer messages for a channel. Here's the key implementation from ConsumerDispatcherChannelBase.cs:

if (_concurrency == 1)
{
 worker = Task.Run(loopStart);
}
else
{
   var tasks = new Task[concurrency];
   for (int i = 0; i < concurrency; i++)
   {
       tasks[i] = Task.Run(loopStart);
   }
   worker = Task.WhenAll(tasks);
}

Answering Your Questions

1. Setting it to 1 vs connection-level fallback

• Setting ConsumerDispatchConcurrency = 1 on a channel does NOT use a shared connection-level dispatcher
• It creates a dedicated single worker task for that specific channel
• The value 1 is the default (Constants.DefaultConsumerDispatchConcurrency = 1)

2. Getting a dedicated dispatch thread per channel

• Any value (including 1) gives you dedicated worker task(s) for that channel
• Each channel gets its own AsyncConsumerDispatcher instance with its own worker tasks
• There is no shared dispatcher across channels - each channel is completely independent

3. Setting it to 2

• Creates 2 dedicated worker tasks for that channel only
• These tasks run concurrently, processing messages from that channel's internal queue
• Other channels are completely unaffected - they have their own separate dispatchers

Channel-Level vs Connection-Level

The connection-level setting is just a default value that channels inherit. From CreateChannelOptions.cs:

internal ushort InternalConsumerDispatchConcurrency
{
   get
   {
       if (ConsumerDispatchConcurrency is not null)
       {
           return ConsumerDispatchConcurrency.Value;  // Channel-level override
       }
       if (_connectionConfigConsumerDispatchConcurrency is not null)
       {
           return connectionConfigConsumerDispatchConcurrency.Value;  // Connection default
       }
       return Constants.DefaultConsumerDispatchConcurrency;  // Global default (1)
   }
}

Key Takeaways

1. Each channel has its own dispatcher - there's no sharing between channels
2. Concurrency = 1 means one dedicated worker task for that channel
3. Concurrency > 1 means multiple worker tasks processing messages concurrently for that channel
4. The connection-level ConsumerDispatchConcurrency is merely a default that channels use if you don't override it per-channel
5. All worker tasks read from a shared System.Threading.Channels.Channel<WorkStruct> queue within each dispatcher

[michaelklishin]

The question surprisingly does not mention the key consideration around this setting: concurrency and shared mutable state access. So I will mention it just in case.

Concurrent dispatch sounds great until you realize that it's on you to synchronize access to any shared mutable state the consumer handlers may use, such as a shared data store connection.

In the Java client, and IIRC older series of this client, there is/was an abstraction that provides both concurrent dispatch across all connection channels and sequential dispatch of messages on a given channel: WorkPool, which hasn't changed much since 2010 because there are no reasons to change it.

[Oliver-Piorun]

Thanks for clarifying! :)