Блог

Я разрабатываю метод использования очереди и тестирую производительность с несколькими потребителями.

В исследованиях рекомендуется увеличивать потребителей, чтобы увеличить пропускную способность очереди.

Я выполнил следующие тесты.

Очередь только с 1 потребителем, вот где я видел самую высокую скорость передачи.

В очереди с 1 каналом и более чем 5 потребителями скорость захвата упала.

Очередь с 1 потребителем на канал, тот же результат, что и раньше.

Я работаю только с 1 работником, чтобы создать все экземпляры потребителей, следует ли использовать более одного работника для каждого потребителя, чтобы обеспечить производительность?

Есть ли у вас подходящая конфигурация для повышения производительности нескольких потребителей?

Следуйте моему методу

  public Task ConsumerRabbitMQAsync<T>(string queeue, int consumerNumber)
        {
            var manualResetEvent = new ManualResetEvent(false);
            manualResetEvent.Reset();

            try
            {
                CreateConnection();
               

                for (int i = 0; i < consumerNumber; i  )
                {
                    var channel = _connection.CreateModel();

                    channel.QueueDeclare(queue: queeue,
                                     durable: false,
                                     exclusive: false,
                                     autoDelete: false,
                                     arguments: null);


                    ConsumerQueeueAsync<T>(queeue, channel);
                }

                manualResetEvent.WaitOne();

            }
            catch (Exception)
            {
                throw;
            }

            return Task.CompletedTask;
        }

        private async void ConsumerQueeueAsync<T>(string queeue, IModel channel)
        {
            var consumer = new AsyncEventingBasicConsumer(channel);

            consumer.Received  = async (model, ea) =>
            {
                try
                {
                    var body = ea.Body;

                    var message = Encoding.UTF8.GetString(body.Span);

                    SendMessage<T>(message);

                    channel.BasicAck(ea.DeliveryTag, false);

                    await Task.Yield();

                }
                catch (Exception ex)
                {
                    channel.BasicNack(ea.DeliveryTag, false, true);
                }
            };

            channel.BasicConsume(queue: queeue,
                                 autoAck: false,
                                 consumer: consumer);

        }