.Net Core RabbitMQ/Masstransit 同一应用程序中每个可配置线程数一个消费者

Posted 2023-03-31

【中文标题】.Net Core RabbitMQ/Masstransit 同一应用程序中每个可配置线程数一个消费者

【英文标题】：.Net Core RabbitMQ/Masstransit one consumer per configurable number of threads in same app

【发布时间】：2021-10-31 13:24:15

【问题描述】：

我可以创建一个workerservices来创建和配置n个RabbitMQ消费者，每个消费者都是自己的线程，用于应用内负载平衡。

我是 Masstransit 的新手，我不知道如何配置我想要的方式。我发现这段代码在 configureservices 中使用消息类型队列：

   // 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.ConfigureConsumer<BasketCheckoutConsumer>(ctx);
                );
            );
        );
        services.AddMassTransitHostedService();

        // General Configuration
        services.AddScoped<BasketCheckoutConsumer>();

问题是，如果我运行从 0 到 n-1 的循环，运行这条线 services.AddScoped<BasketCheckoutConsumer>();

这个箱子会产生 n 个消费者吗？也就是说，消费者从同一个队列中获取消息。

我的方向是否正确？

【参考方案1】：

MassTransit 中的消费者不会连续使用线程。消费者也在单个消息处理的范围内被实例化，因此在涉及消费者实例的数量时谈论线程是没有意义的。

MassTransit 在 IO 方面是完全异步的，它会实例化同一消费者的多个实例来处理负载。您可以通过在配置中设置ConcurtrentMessageLimit 来指定给定接收端点需要多少并发消费者，因为它是described in the docs。还要记住adjust the prefetch count，这样您才能真正获得所需的并发级别。

【讨论】：

在接收端点上设置 ConcurrentMessageLimit 优于添加 UseConcurrencyLimit。

谢谢@ChrisPatterson，我会更新答案

@ChrisPatterson 顺便说一句，文档说“ConcurrentMessageLimit 仅传递给支持它的传输（当前为 Azure 服务总线）”

是的，我需要将文档更新为match the release notes。

我阅读了更多内容，发现我需要使用单个消费者类配置“竞争消费者模式”的方法。我没有找到如何在启动类中配置它的单个示例，您能提供一个示例吗？