.Net Core下Masstransit多RabbitMQ消费者同队列消费配置问题

结论先行 #

你的实现思路是错误的，循环执行services.AddScoped<BasketCheckoutConsumer>()完全无法生成N个从同一队列拉取消息的消费者。

错误原因 #

services.AddScoped<T>()是.NET依赖注入容器的服务注册方法：

• 重复注册同类型的Scoped服务不会产生多个实例，DI容器只会保留该类型的服务注册规则，每次解析该类型时只会生成一个实例，用完随作用域销毁
• 该注册操作和MassTransit运行时启动的消费线程数、消费者实例数没有任何关联，完全不影响消费端的并行配置

另外额外提醒：你现有代码中的services.AddScoped<BasketCheckoutConsumer>()是冗余的，调用config.AddConsumer<BasketCheckoutConsumer>()时，MassTransit已经自动完成了消费者的DI注册，不需要手动重复注册。

正确实现方案 #

根据你的需求场景，有两种主流实现方式：

场景1：普通高并发消费场景（90%以上适用） #

仅需要提升单进程内同队列的消费并行能力，直接配置接收端点的ConcurrentMessageLimit参数即可，值设为你需要的N，MassTransit会自动维护N个并行消费线程，RabbitMQ会自动将消息轮询分发到空闲线程，实现应用内负载均衡。
修改后的配置代码如下：

// MassTransit-RabbitMQ Configuration
services.AddMassTransit(config => {

    config.AddConsumer<BasketCheckoutConsumer>();

    config.UsingRabbitMq((ctx, cfg) => {
        cfg.Host(Configuration["EventBusSettings:HostAddress"]);
        cfg.UseHealthCheck(ctx);

        cfg.ReceiveEndpoint(EventBusConstants.BasketCheckoutQueue, c => {
            c.ConcurrentMessageLimit = N; // 替换为你需要的并发数，比如8就是同时最多处理8条消息
            c.PrefetchCount = (ushort)(N * 2); // 预取数建议设为并发数的1~2倍，优化消费性能
            c.ConfigureConsumer<BasketCheckoutConsumer>(ctx);
        });
    });
});
services.AddMassTransitHostedService();

场景2：需要完全独立的多个消费者实例 #

如果要求每个消费者有独立的连接、独立的异常策略等完全隔离的配置，可以循环创建N个同名的接收端点，每个端点绑定同一个队列即可：

config.UsingRabbitMq((ctx, cfg) => {
    cfg.Host(Configuration["EventBusSettings:HostAddress"]);
    cfg.UseHealthCheck(ctx);

    // 循环生成N个独立消费者，绑定同一个队列
    for(int i = 0; i < N; i++)
    {
        cfg.ReceiveEndpoint(EventBusConstants.BasketCheckoutQueue, c => {
            c.ConfigureConsumer<BasketCheckoutConsumer>(ctx);
            // 可单独配置每个端点的参数
        });
    }
});

注意：该方式会增加与RabbitMQ的连接数，消耗更多系统资源，没有特殊需求优先使用第一种方案。

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