Masstransit JobConsumer在K8s RabbitMQ环境下并发限制异常求助

基于JobConsumersSample测试MassTransit的JobConsumer组件时，本地使用masstransit/rabbitmq镜像能正常遵循配置的并发限制；但部署到K8s中的RabbitMQ（版本3.9.16，搭配rabbitmq-delayed-message-exchange V3.9.0）时，首分钟仅处理1个任务，之后稳定每分钟处理2个，和配置的20并发限制不符。怀疑是RabbitMQ配置问题，但不知道从何入手。

相关配置如下：

ConsumerDefinition代码 #

public class ExtractDocumentJobConsumerConfiguration : ConsumerDefinition<ExtractDocumentJobConsumer>
{
    protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator,
  IConsumerConfigurator<ExtractDocumentJobConsumer> consumerConfigurator)
    {
        consumerConfigurator.Options<JobOptions<ExtractDocument>>(options =>
            options
                .SetRetry(r => r.Interval(3, TimeSpan.FromSeconds(30)))
                .SetJobTimeout(TimeSpan.FromMinutes(10))
                .SetConcurrentJobLimit(20));
    }
}

Program.cs代码 #

builder.Services
        .AddMassTransit(
            x =>
            {
                x.AddDelayedMessageScheduler();

                x.AddConsumer<ExtractDocumentJobConsumer, ExtractDocumentJobConsumerConfiguration>();
                //.Endpoint(e => e.Name = "extraction-job-queue");

                x.AddConsumer<TrackAnalysisJobConsumer>();

                x.AddSagaRepository<JobSaga>().InMemoryRepository();
                x.AddSagaRepository<JobTypeSaga>().InMemoryRepository();
                x.AddSagaRepository<JobAttemptSaga>().InMemoryRepository();

                x.SetKebabCaseEndpointNameFormatter();

                x.UsingRabbitMq(
                    (context, cfg) =>
                    {
                        cfg.Host(
                            rabbitMQConfiguration.HostName,
                            rabbitMQConfiguration.VirtualHost,
                            h =>
                            {
                                h.Username(rabbitMQConfiguration.UserName);
                                h.Password(rabbitMQConfiguration.Password);
                            });

                        cfg.UseDelayedMessageScheduler();

                        var options = new ServiceInstanceOptions()
                            .SetEndpointNameFormatter(context.GetService<IEndpointNameFormatter>() ?? KebabCaseEndpointNameFormatter.Instance);

                        cfg.ServiceInstance(
                            options,
                            instance =>
                            {
                                instance.ConfigureJobServiceEndpoints(
                                    js =>
                                    {
                                        js.SagaPartitionCount = 1;
                                        js.FinalizeCompleted = true;
                                        js.ConfigureSagaRepositories(context);
                                    });
                                //instance.InstanceEndpointConfigurator.ConcurrentMessageLimit = 20;
                                instance.ConfigureEndpoints(
                                    context,
                                    f => f.Include<ExtractDocumentJobConsumer>());
                            });

                        // Configure the remaining consumers
                        cfg.ConfigureEndpoints(context);
                    });
            });

builder.Services
  .AddOptions<MassTransitHostOptions>()
  .Configure(
      options =>
      {
          options.WaitUntilStarted = true;
          options.StartTimeout = TimeSpan.FromSeconds(45);
          options.StopTimeout = TimeSpan.FromSeconds(45);
      });

注：必须注释掉.Endpoint(e => e.Name = "extraction-job-queue")，否则会提示端点已存在。

• 检查RabbitMQ的prefetch count设置：进入RabbitMQ管理后台，找到对应作业队列，查看Prefetch Count是否≥20。MassTransit通常会根据并发限制自动设置该值，但如果RabbitMQ有全局或队列级别的配置覆盖，会导致无法拉取足够消息。
• 调整Job Service的Saga分区数：当前设置js.SagaPartitionCount = 1，单分区的JobTypeSaga可能成为并发瓶颈。尝试将分区数调整为20或更高，确保Saga能及时分发任务。
• 验证延迟插件状态：查看RabbitMQ日志，搜索delayed_message_exchange相关内容，确认插件已正确启用且无运行异常。
• 开启MassTransit Debug日志：观察任务分发过程中的日志，检查是否有Job accepted、Job dispatched等关键日志，确认是任务分发不足还是消费者接收异常。
• 检查K8s Pod资源限制：确认Pod的CPU、内存请求和限制是否足够支撑20个并发任务，资源不足会导致消费者无法同时处理多个作业。
• 查看队列消费者状态：在RabbitMQ管理后台查看作业队列的consumer count和active consumers，确保有足够的消费者实例且处于活跃状态。
• 排查端点配置冲突：尝试通过ConfigureJobServiceEndpoints显式指定作业相关端点的并发配置，避免自动创建的端点出现设置遗漏。

内容的提问来源于stack exchange，提问作者jweaver