Allow the dispatcher concurrency to be overridden per channel

danielmarbach · danielmarbach · commit df62f863188b · 2024-09-11T11:21:00.000+02:00
diff --git a/projects/RabbitMQ.Client/PublicAPI.Shipped.txt b/projects/RabbitMQ.Client/PublicAPI.Shipped.txt
@@ -826,7 +826,7 @@ virtual RabbitMQ.Client.TcpClientAdapter.ReceiveTimeout.set -> void
 ~RabbitMQ.Client.IChannel.WaitForConfirmsAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<bool>
 ~RabbitMQ.Client.IChannel.WaitForConfirmsOrDieAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task
 ~RabbitMQ.Client.IConnection.CloseAsync(ushort reasonCode, string reasonText, System.TimeSpan timeout, bool abort, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task
-~RabbitMQ.Client.IConnection.CreateChannelAsync(System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel>
+~RabbitMQ.Client.IConnection.CreateChannelAsync(int? dispatchConsumerConcurrency = null, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IChannel>
 ~RabbitMQ.Client.IConnection.UpdateSecretAsync(string newSecret, string reason, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task
 ~RabbitMQ.Client.IConnectionFactory.CreateConnectionAsync(string clientProvidedName, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IConnection>
 ~RabbitMQ.Client.IConnectionFactory.CreateConnectionAsync(System.Collections.Generic.IEnumerable<RabbitMQ.Client.AmqpTcpEndpoint> endpoints, string clientProvidedName, System.Threading.CancellationToken cancellationToken = default(System.Threading.CancellationToken)) -> System.Threading.Tasks.Task<RabbitMQ.Client.IConnection>
diff --git a/projects/RabbitMQ.Client/client/api/IConnection.cs b/projects/RabbitMQ.Client/client/api/IConnection.cs
@@ -234,8 +234,17 @@ Task CloseAsync(ushort reasonCode, string reasonText, TimeSpan timeout, bool abo
         /// <summary>
         /// Asynchronously create and return a fresh channel, session, and channel.
         /// </summary>
+        /// <param name="dispatchConsumerConcurrency">
+        /// Set to a value greater than one to enable concurrent processing. For a concurrency greater than one <see cref="IAsyncBasicConsumer"/>
+        /// will be offloaded to the worker thread pool so it is important to choose the value for the concurrency wisely to avoid thread pool overloading.
+        /// <see cref="IAsyncBasicConsumer"/> can handle concurrency much more efficiently due to the non-blocking nature of the consumer.
+        ///
+        /// Defaults to <see cref="IConnectionFactory.ConsumerDispatchConcurrency"/>.
+        /// 
+        /// For concurrency greater than one this removes the guarantee that consumers handle messages in the order they receive them.
+        /// In addition to that consumers need to be thread/concurrency safe.
+        /// </param>
         /// <param name="cancellationToken">Cancellation token</param>
-        Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default);
-
+        Task<IChannel> CreateChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default);
     }
 }
diff --git a/projects/RabbitMQ.Client/client/framing/Channel.cs b/projects/RabbitMQ.Client/client/framing/Channel.cs
@@ -38,7 +38,7 @@ namespace RabbitMQ.Client.Framing.Impl
 {
     internal class Channel : ChannelBase
     {
-        public Channel(ConnectionConfig config, ISession session) : base(config, session)
+        public Channel(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null) : base(config, session, dispatchConsumerConcurrency)
         {
         }
 
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringChannel.cs
@@ -160,7 +160,7 @@ internal async Task<bool> AutomaticallyRecoverAsync(AutorecoveringConnection con
 
             _connection = conn;
 
-            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(cancellationToken)
+            RecoveryAwareChannel newChannel = await conn.CreateNonRecoveringChannelAsync(cancellationToken: cancellationToken)
                 .ConfigureAwait(false);
             newChannel.TakeOver(_innerChannel);
 
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.Recovery.cs
@@ -296,7 +296,7 @@ private async ValueTask RecoverExchangesAsync(IConnection connection,
             {
                 try
                 {
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))
+                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
                     {
                         await recordedExchange.RecoverAsync(ch, cancellationToken)
                             .ConfigureAwait(false);
@@ -348,7 +348,7 @@ private async Task RecoverQueuesAsync(IConnection connection,
                 try
                 {
                     string newName = string.Empty;
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))
+                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
                     {
                         newName = await recordedQueue.RecoverAsync(ch, cancellationToken)
                             .ConfigureAwait(false);
@@ -459,7 +459,7 @@ private async ValueTask RecoverBindingsAsync(IConnection connection,
             {
                 try
                 {
-                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))
+                    using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
                     {
                         await binding.RecoverAsync(ch, cancellationToken)
                             .ConfigureAwait(false);
diff --git a/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs b/projects/RabbitMQ.Client/client/impl/AutorecoveringConnection.cs
@@ -173,10 +173,10 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer
 
         public IProtocol Protocol => Endpoint.Protocol;
 
-        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(CancellationToken cancellationToken)
+        public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default)
         {
             ISession session = InnerConnection.CreateSession();
-            var result = new RecoveryAwareChannel(_config, session);
+            var result = new RecoveryAwareChannel(_config, session, dispatchConsumerConcurrency);
             return (RecoveryAwareChannel)await result.OpenAsync(cancellationToken).ConfigureAwait(false);
         }
 
@@ -239,10 +239,10 @@ await CloseInnerConnectionAsync()
             }
         }
 
-        public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
+        public async Task<IChannel> CreateChannelAsync(int? dispatchConsumerConcurrency = null,CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
-            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cancellationToken)
+            RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(dispatchConsumerConcurrency, cancellationToken)
                 .ConfigureAwait(false);
             AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel);
             await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)
diff --git a/projects/RabbitMQ.Client/client/impl/ChannelBase.cs b/projects/RabbitMQ.Client/client/impl/ChannelBase.cs
@@ -73,10 +73,10 @@ internal abstract class ChannelBase : IChannel, IRecoverable
 
         internal readonly IConsumerDispatcher ConsumerDispatcher;
 
-        protected ChannelBase(ConnectionConfig config, ISession session)
+        protected ChannelBase(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null)
         {
             ContinuationTimeout = config.ContinuationTimeout;
-            ConsumerDispatcher = new AsyncConsumerDispatcher(this, config.DispatchConsumerConcurrency);
+            ConsumerDispatcher = new AsyncConsumerDispatcher(this, dispatchConsumerConcurrency.GetValueOrDefault(config.DispatchConsumerConcurrency));
             Action<Exception, string> onException = (exception, context) =>
                 OnCallbackException(CallbackExceptionEventArgs.Build(exception, context));
             _basicAcksWrapper = new EventingWrapper<BasicAckEventArgs>("OnBasicAck", onException);
diff --git a/projects/RabbitMQ.Client/client/impl/Connection.cs b/projects/RabbitMQ.Client/client/impl/Connection.cs
@@ -253,11 +253,11 @@ await CloseAsync(ea, true,
             }
         }
 
-        public Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)
+        public Task<IChannel> CreateChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default)
         {
             EnsureIsOpen();
             ISession session = CreateSession();
-            var channel = new Channel(_config, session);
+            var channel = new Channel(_config, session, dispatchConsumerConcurrency);
             return channel.OpenAsync(cancellationToken);
         }
 
diff --git a/projects/RabbitMQ.Client/client/impl/RecoveryAwareChannel.cs b/projects/RabbitMQ.Client/client/impl/RecoveryAwareChannel.cs
@@ -37,7 +37,7 @@ namespace RabbitMQ.Client.Impl
 {
     internal sealed class RecoveryAwareChannel : Channel
     {
-        public RecoveryAwareChannel(ConnectionConfig config, ISession session) : base(config, session)
+        public RecoveryAwareChannel(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null) : base(config, session, dispatchConsumerConcurrency)
         {
             ActiveDeliveryTagOffset = 0;
             MaxSeenDeliveryTag = 0;

Original file line number	Diff line number	Diff line change
`@@ -38,7 +38,7 @@ namespace RabbitMQ.Client.Framing.Impl`
`38`	`38`	`{`
`39`	`39`	`internal class Channel : ChannelBase`
`40`	`40`	`{`
`41`		`- public Channel(ConnectionConfig config, ISession session) : base(config, session)`
	`41`	`+ public Channel(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null) : base(config, session, dispatchConsumerConcurrency)`
`42`	`42`	`{`
`43`	`43`	`}`
`44`	`44`
Original file line number	Diff line number	Diff line change
`@@ -296,7 +296,7 @@ private async ValueTask RecoverExchangesAsync(IConnection connection,`
`296`	`296`	`{`
`297`	`297`	`try`
`298`	`298`	`{`
`299`		`- using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))`
	`299`	`+ using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))`
`300`	`300`	`{`
`301`	`301`	`await recordedExchange.RecoverAsync(ch, cancellationToken)`
`302`	`302`	`.ConfigureAwait(false);`
`@@ -348,7 +348,7 @@ private async Task RecoverQueuesAsync(IConnection connection,`
`348`	`348`	`try`
`349`	`349`	`{`
`350`	`350`	`string newName = string.Empty;`
`351`		`- using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))`
	`351`	`+ using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))`
`352`	`352`	`{`
`353`	`353`	`newName = await recordedQueue.RecoverAsync(ch, cancellationToken)`
`354`	`354`	`.ConfigureAwait(false);`
`@@ -459,7 +459,7 @@ private async ValueTask RecoverBindingsAsync(IConnection connection,`
`459`	`459`	`{`
`460`	`460`	`try`
`461`	`461`	`{`
`462`		`- using (IChannel ch = await connection.CreateChannelAsync(cancellationToken).ConfigureAwait(false))`
	`462`	`+ using (IChannel ch = await connection.CreateChannelAsync(cancellationToken: cancellationToken).ConfigureAwait(false))`
`463`	`463`	`{`
`464`	`464`	`await binding.RecoverAsync(ch, cancellationToken)`
`465`	`465`	`.ConfigureAwait(false);`
Original file line number	Diff line number	Diff line change
`@@ -173,10 +173,10 @@ public event EventHandler<RecoveringConsumerEventArgs> RecoveringConsumer`
`173`	`173`
`174`	`174`	`public IProtocol Protocol => Endpoint.Protocol;`
`175`	`175`
`176`		`- public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(CancellationToken cancellationToken)`
	`176`	`+ public async ValueTask<RecoveryAwareChannel> CreateNonRecoveringChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default)`
`177`	`177`	`{`
`178`	`178`	`ISession session = InnerConnection.CreateSession();`
`179`		`- var result = new RecoveryAwareChannel(_config, session);`
	`179`	`+ var result = new RecoveryAwareChannel(_config, session, dispatchConsumerConcurrency);`
`180`	`180`	`return (RecoveryAwareChannel)await result.OpenAsync(cancellationToken).ConfigureAwait(false);`
`181`	`181`	`}`
`182`	`182`
`@@ -239,10 +239,10 @@ await CloseInnerConnectionAsync()`
`239`	`239`	`}`
`240`	`240`	`}`
`241`	`241`
`242`		`- public async Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)`
	`242`	`+ public async Task<IChannel> CreateChannelAsync(int? dispatchConsumerConcurrency = null,CancellationToken cancellationToken = default)`
`243`	`243`	`{`
`244`	`244`	`EnsureIsOpen();`
`245`		`- RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(cancellationToken)`
	`245`	`+ RecoveryAwareChannel recoveryAwareChannel = await CreateNonRecoveringChannelAsync(dispatchConsumerConcurrency, cancellationToken)`
`246`	`246`	`.ConfigureAwait(false);`
`247`	`247`	`AutorecoveringChannel channel = new AutorecoveringChannel(this, recoveryAwareChannel);`
`248`	`248`	`await RecordChannelAsync(channel, channelsSemaphoreHeld: false, cancellationToken: cancellationToken)`
Original file line number	Diff line number	Diff line change
`@@ -253,11 +253,11 @@ await CloseAsync(ea, true,`
`253`	`253`	`}`
`254`	`254`	`}`
`255`	`255`
`256`		`- public Task<IChannel> CreateChannelAsync(CancellationToken cancellationToken = default)`
	`256`	`+ public Task<IChannel> CreateChannelAsync(int? dispatchConsumerConcurrency = null, CancellationToken cancellationToken = default)`
`257`	`257`	`{`
`258`	`258`	`EnsureIsOpen();`
`259`	`259`	`ISession session = CreateSession();`
`260`		`- var channel = new Channel(_config, session);`
	`260`	`+ var channel = new Channel(_config, session, dispatchConsumerConcurrency);`
`261`	`261`	`return channel.OpenAsync(cancellationToken);`
`262`	`262`	`}`
`263`	`263`
Original file line number	Diff line number	Diff line change
`@@ -37,7 +37,7 @@ namespace RabbitMQ.Client.Impl`
`37`	`37`	`{`
`38`	`38`	`internal sealed class RecoveryAwareChannel : Channel`
`39`	`39`	`{`
`40`		`- public RecoveryAwareChannel(ConnectionConfig config, ISession session) : base(config, session)`
	`40`	`+ public RecoveryAwareChannel(ConnectionConfig config, ISession session, int? dispatchConsumerConcurrency = null) : base(config, session, dispatchConsumerConcurrency)`
`41`	`41`	`{`
`42`	`42`	`ActiveDeliveryTagOffset = 0;`
`43`	`43`	`MaxSeenDeliveryTag = 0;`