我们使用的是带RabbitMQ的Mastransit 8.0.2（3.8.1 Erlang 22.1.5）和.Net6，消息从TCPClient应用发布，消息发布时间逐渐增加，发布一条消息最多需要30分钟，所有消息都在后台TCP接收方服务发布，数据速率最快可达60-在TCP客户端接收到70条消息，并使用Mastransit异步发布到Rabbit。

• DataMessage消费者部署在3个不同的服务器上，总共5个消费者的消费者利用率为100%（RabbitMQ Portal）。
• 所有使用者的预取计数均为100（我们尝试了1、5、16、50、100，预取计数结果相同）
• 此队列上的消息发布和确认速率平均为10/s。（RabbitMQ服务器配置4核CPU，16 GB RAM，磁盘I/O 6400）

我们已经尝试了多种配置，包括预取计数以及增加和减少使用者。我们还查看了服务器CPU/内存/网络利用率，它们的平均值都低于50%总线在应用程序启动时启动。

public static void AddServiceBus(this IServiceCollection services, IConfiguration configuration, int prefetchCount = 0, params Type[] consumers)
        {
            services.AddMassTransit(x =>
            {
                if (consumers != null && consumers.Any())
                {
                    x.AddConsumers(consumers);
                }
                x.UsingRabbitMq((context, configurator) =>
                {
                    var rabbitMqSettings = configuration.GetSection(nameof(RabbitMqConfiguration)).Get<RabbitMqConfiguration>();
                    configurator.Host(rabbitMqSettings.Host, d =>
                    {
                        d.Username(rabbitMqSettings.Username);
                        d.Password(rabbitMqSettings.Password);
                    });
                    configurator.ConfigureEndpoints(context);
                    configurator.UseRetry(b =>
                    {
                        b.Immediate(3);
                    });

                    if (prefetchCount > 0)
                        configurator.PrefetchCount = prefetchCount;
                });
                x.Configure<MassTransitHostOptions>(options =>
                {
                    options.WaitUntilStarted = true;
                    options.StartTimeout = TimeSpan.FromSeconds(30);
                    options.StopTimeout = TimeSpan.FromMinutes(1);

                });
            });
        }

private async Task<ErrorCode> PublishDataAsync(BaseData Data, string _messageGuid)
        {
            try
            {
                using var scope = _serviceProviderFactory.CreateScope();
                var publishEndpoint = scope.ServiceProvider.GetRequiredService<IPublishEndpoint>();

                var DataMessage = new DataMessage(Data);
                await _publishEndpoint.Publish(DataMessage);
                _logger.LogInformation("{messgeguid} Data Published to MassTransit", _messageGuid);
                return ErrorCodes.SUCCESS;
            }
            catch (Exception e)
            {
                _logger.LogError(e, $"Message could not be published. {JsonConvert.SerializeObject(Data)}");
            }

        }

我们正在尝试增加消息发布速率。我们的发布消息大小为平均2KB。