Ограничение пропускной способности Masstransit RPC (RabbitMq) - PullRequest
Новая
       62

Ограничение пропускной способности Masstransit RPC (RabbitMq)

0 голосов
/ 25 октября 2018

Мы используем Masstransit с RabbitMq для создания RPC из одного компонента нашей системы в другие.

Недавно мы столкнулись с ограничением пропускной способности на стороне клиента, измеряемым около 80 завершенных ответов в секунду.

При попытке выяснить, в чем проблема, я обнаружил, что запросы RPC-сервером быстро обрабатывались, затем ответы помещались в очередь обратного вызова, а затем скорость обработки в очереди составляла 80 М \ с

Этопредел только на стороне клиента.Запуск другого процесса того же клиентского приложения на том же компьютере удваивает пропускную способность запросов на стороне сервера, но затем я вижу, что используются две очереди обратного вызова, заполненные сообщениями, каждая с одинаковыми 80 M \ s

Мыиспользуется один экземпляр IBus 

builder.Register(c =>
{
    var busSettings = c.Resolve<RabbitSettings>();
    var busControl = MassTransitBus.Factory.CreateUsingRabbitMq(cfg =>
        {
            var host = cfg.Host(new Uri(busSettings.Host), h =>
            {
                h.Username(busSettings.Username);
                h.Password(busSettings.Password);
            });

            cfg.UseSerilog();

            cfg.Send<IProcessorContext>(x =>
            {
               x.UseCorrelationId(context => context.Scope.CommandContext.CommandId);
            });

    }
);

return busControl;
})
.As<IBusControl>()
.As<IBus>()
.SingleInstance();

Логика отправки выглядит следующим образом: 

var busResponse = await _bus.Request<TRequest, TResult>(
                destinationAddress: _settings.Host.GetServiceUrl<TCommand>(queueType),
                message: commandContext,
                cancellationToken: default(CancellationToken),
                timeout: TimeSpan.FromSeconds(_settings.Timeout),
                callback: p => { p.WithPriority(priority); });

Кто-нибудь сталкивался с подобной проблемой?Я предполагаю, что в логике отправки ответа есть какой-то программный предел.Это может быть максимальный размер пула потоков или размер буфера, а также счетчик предварительной выборки очереди ответов.Я пытался поиграть с размером пула потоков .Net, но ничего не помогло.

Я новичок в Masstransit и буду признателен за любую помощь в решении моей проблемы.Надеюсь, что это может быть исправлено в конфигурации

1 Ответ

0 голосов
/ 25 октября 2018

Есть несколько вещей, которые вы можете попытаться оптимизировать.Я бы также предложил проверить MassTransit-Benchmark и запустить его в своей среде - это даст вам представление о возможной пропускной способности вашего брокера.Он позволяет вам настраивать такие параметры, как количество предварительных выборок, параллелизм и т. Д., Чтобы увидеть, как они влияют на ваши результаты.

Кроме того, я бы предложил использовать один из клиентов запросов, чтобы уменьшить настройки для каждого запроса / ответа.Например, создайте клиент запроса один раз, а затем используйте этот же клиент для каждого запроса.

var serviceUrl = yourMethodToGetIt<TRequest>(...); var client = Bus.CreateRequestClient<TRequest>(serviceUrl);

Затем используйте этот экземпляр IRequestClient<TRequest> всякий раз, когда вам нужно выполнить запрос.

Response<Value> response = await client.GetResponse<TResponse>(new Request());

Поскольку вы используете только RPC, я настоятельно рекомендую установить очередь конечной точки приема на недолговечную, чтобы избежать записи запросов RPC на диск.И отрегулируйте счетчик предварительной выборки шины до более высокого значения (больше, чем максимальное число одновременных запросов, которое вы можете иметь в 2 раза), чтобы гарантировать, что ответы всегда доставляются непосредственно ожидающему получателю ответов (это внутренняя вещь в том, как RabbitMQ доставляет сообщения).

var busControl = Bus.Factory.CreateUsingRabbitMq(cfg => { cfg.PrefetchCount = 1000; }

...