Refactor RabbitMQ consumer logic for clarity and QoS

Refactored `HandleBasicDeliverAsync` in `RabbitMqBasicConsumer` to simplify `safeBody` handling and eliminate redundant variable usage. Updated to pass `body` directly to `BasicDeliverEventArgs` and `TransportMessage`.

Enhanced `Listening` method in `RabbitMqConsumerClient` to use `BasicQosOptions` from `_rabbitMqOptions` for more flexible QoS settings. Defaulted `prefetchCount` to `_groupConcurrent` or 1 for better configurability.

Author: yang-xiaodong

src/DotNetCore.CAP.RabbitMQ/RabbitMQBasicConsumer.cs

@@ -45,12 +45,12 @@ public override async Task HandleBasicDeliverAsync(string consumerTag, ulong del
         string routingKey, IReadOnlyBasicProperties properties, ReadOnlyMemory<byte> body,
         CancellationToken cancellationToken = default)
     {
-        var safeBody = _usingTaskRun ? body.ToArray() : body;
-
         if (_usingTaskRun)
         {
             await _semaphore.WaitAsync(cancellationToken);
 
+            var safeBody = body.ToArray();
+
             _ = Task.Run(Consume, cancellationToken).ConfigureAwait(false);
         }
         else
@@ -76,15 +76,15 @@ Task Consume()
             if (_customHeadersBuilder != null)
             {
                 var e = new BasicDeliverEventArgs(consumerTag, deliveryTag, redelivered, exchange, routingKey,
-                    properties, safeBody);
+                    properties, body);
                 var customHeaders = _customHeadersBuilder(e, _serviceProvider);
                 foreach (var customHeader in customHeaders)
                 {
                     headers[customHeader.Key] = customHeader.Value;
                 }
             }
 
-            var message = new TransportMessage(headers, safeBody);
+            var message = new TransportMessage(headers, body);
 
             return _msgCallback(message, deliveryTag);
         }

src/DotNetCore.CAP.RabbitMQ/RabbitMQConsumerClient.cs

@@ -60,13 +60,14 @@ public void Listening(TimeSpan timeout, CancellationToken cancellationToken)
     {
         Connect().GetAwaiter().GetResult();
 
-        if (_groupConcurrent > 0)
+        if (_rabbitMqOptions.BasicQosOptions != null)
         {
-            _channel!.BasicQosAsync(prefetchSize: 0, prefetchCount: _groupConcurrent, global: false, cancellationToken).GetAwaiter().GetResult(); 
+            _channel!.BasicQosAsync(0, _rabbitMqOptions.BasicQosOptions.PrefetchCount, _rabbitMqOptions.BasicQosOptions.Global, cancellationToken).GetAwaiter().GetResult();
         }
-        else if (_rabbitMqOptions.BasicQosOptions != null)
+        else
         {
-            _channel!.BasicQosAsync(0, _rabbitMqOptions.BasicQosOptions.PrefetchCount, _rabbitMqOptions.BasicQosOptions.Global, cancellationToken).GetAwaiter().GetResult();
+            ushort prefetch = _groupConcurrent > 0 ? _groupConcurrent : (ushort)1;
+            _channel!.BasicQosAsync(prefetchSize: 0, prefetchCount: prefetch, global: false, cancellationToken).GetAwaiter().GetResult();
         }
 
         _consumer = new RabbitMqBasicConsumer(_channel!, _groupConcurrent, _queueName, OnMessageCallback!, OnLogCallback!,