lower performance in publiching batches in the v7 version

[GhoufranGhazaly]

I upgraded the Rabbitmq client version from 6.8 to 7.1.2 and I faced low performance related to publich batches, I followed Publishing Messages in Batches in this link https://www.rabbitmq.com/tutorials/tutorial-seven-dotnet

[michaelklishin]

@GhoufranGhazaly we cannot suggest anything without an executable way to reproduce. We do not guess in this community, and we won't guess how exactly you "publish in batches".

[GhoufranGhazaly]

before upgrade this how I was publishing batches

public async Task<bool> PublishBatchAsync(List<RMQMessage> rmqMessages)
 {
     if (rmqMessages.Count == 0)
         return false;

     IModel rabbitChannel = null!;
     IBasicPublishBatch batch = null!;

     try
     {
         if ((_connection == null) || !_connection.IsOpen)
         {
             // Requeue Messages to internal batch concurrent queue
             if (!IsBatchMessageInQueue_v2(rmqMessages))
             {
                 _concurrentBatchPublishQueue_v2.Enqueue(rmqMessages);
                 _logger.LogTrace(nameof(RabbitMQPublish), DateTime.Now, string.Empty, $"RMQBatchCount [added to concurrent queue]: {rmqMessages.Count}");
             }

             // Initiate the re-connection
             await createConnection();
         }

         // Create RMQ batch session/channel from existing connection
         rabbitChannel = _connection!.CreateModel();
         rabbitChannel.ConfirmSelect();
         batch = rabbitChannel.CreateBasicPublishBatch();

         ReadOnlyMemory<byte> buffer;
         foreach (var message in rmqMessages)
         {
             buffer = new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes(message.Message!));

             IBasicProperties _properties = rabbitChannel.CreateBasicProperties();
             _properties.Persistent = true;
             _properties.Headers = message.Headers ?? new Dictionary<string, object>();

             string exchange = message.Exchange!;
             string routingKey = message.RoutingKey_or_QueueName!;

             batch.Add(exchange, routingKey, true, _properties, buffer);
         }

         await Task.Run(() =>
         {
             batch.Publish();
             rabbitChannel.WaitForConfirms();
         });

         rmqMessages.Clear();

         if (rabbitChannel.IsOpen)
             rabbitChannel.Close();

         // Re-Process the messages in _concurrentBatchPublishQueue
         await ReProcessConcurrentBatchPublishMessages_v2();

         return true;
     }
     catch (Exception ex)
     {
        _logger.LogError(nameof(RabbitMQPublish),DateTime.Now,ex);
         return false;
     }
     finally
     {
         if (rabbitChannel != null && rabbitChannel.IsOpen)
             rabbitChannel.Close();
     }
 }

And after upgrade rabbitmq client version, this is how I publish batches

  public async Task<bool> PublishBatchAsync(List<RMQMessage> rmqMessages)
  {
      if (rmqMessages.Count == 0)
          return false;
      IChannel rabbitChannel=null;
      try {
      if ((_connection == null) || !_connection.IsOpen)
      {
          // Requeue Messages to internal batch concurrent queue
          if (!IsBatchMessageInQueue_v2(rmqMessages))
          {
              _concurrentBatchPublishQueue_v2.Enqueue(rmqMessages);
              _logger.LogTrace(nameof(RabbitMQPublish), DateTime.Now, string.Empty, $"RMQBatchCount [added to concurrent queue]: {rmqMessages.Count}");
          }

          // Initiate the re-connection
          await createConnection();
      }
          var channelOpts = new CreateChannelOptions(
              publisherConfirmationsEnabled: true,
              publisherConfirmationTrackingEnabled: false,
              outstandingPublisherConfirmationsRateLimiter: new ThrottlingRateLimiter(MAX_OUTSTANDING_CONFIRMS)
          );
       rabbitChannel = await _connection.CreateChannelAsync(channelOpts);

       int batchSize = rmqMessages.Count();// Math.Max(1, MAX_OUTSTANDING_CONFIRMS / 2);

      //var sw = Stopwatch.StartNew();

      var publishTasks = new List<ValueTask>();
      ReadOnlyMemory<byte> buffer;
      foreach (var message in rmqMessages)
      {
          buffer = new ReadOnlyMemory<byte>(Encoding.UTF8.GetBytes(message.Message!));

          var _properties = new BasicProperties
          {
              Persistent = true,
              Headers = message.Headers ?? new Dictionary<string, object>(),

          };
          string exchange = message.Exchange!;
          string routingKey = message.RoutingKey_or_QueueName!;
          ValueTask publishTask = rabbitChannel.BasicPublishAsync(exchange: exchange, routingKey: routingKey, body: buffer, mandatory: true, basicProperties: _properties);
          publishTasks.Add(publishTask);

          await MaybeAwaitPublishes(publishTasks, batchSize);
      }

          // Await any remaining tasks in case message count was not
          // evenly divisible by batch size.
          await MaybeAwaitPublishes(publishTasks, 0);
      
      
      rmqMessages.Clear();
      if (rabbitChannel.IsOpen)
          await rabbitChannel.CloseAsync();
      await ReProcessConcurrentBatchPublishMessages_v2();
      return true;
      }
      catch (Exception ex)
      {
          _logger.LogError(nameof(RabbitMQPublishOld), DateTime.Now, ex);
          return false;
      }
      finally
      {
          if (rabbitChannel != null && rabbitChannel.IsOpen)
              rabbitChannel.CloseAsync();
      }

  }
  public async Task MaybeAwaitPublishes(List<ValueTask> publishTasks, int batchSize)
  {
      if (publishTasks.Count >= batchSize)
      {
      
          foreach (ValueTask pt in publishTasks)
          {
              try
              {
                  await pt;
              }
              catch (Exception ex) /* handle NACKs/returns as exceptions */ 
              {
                  _logger.LogError(nameof(RabbitMQPublish), DateTime.Now, ex);
              }
          }
          publishTasks.Clear();
      }
  }

[GhoufranGhazaly]

the code using the library before upgrading takes 4312 ms to publish 200-000 messages and the batch size is 1000, the code after upgrade takes around 74913 ms for same number of messages and same batch size.

[michaelklishin]

By "executable version" I mean an example in a public repository that can be cloned, built and run.

Profilers cannot be applied to a comment on GitHub but they can be applied to an executing process.

[lukebakken]

This is the problem:

int batchSize = rmqMessages.Count();// Math.Max(1, MAX_OUTSTANDING_CONFIRMS / 2);

As soon as your list of messages pending confirmations exceeds half the outstanding confirms count, the client library will rate-limit your publishes. This is to prevent your application from overwhelming the broker.

Please modify your code to publish in batches in exactly the same manner as the example. If you need assistance, provide a GitHub repository with code I can clone, compile, and run. I can then more easily assist you.