端点与消费者的ConcurrentMessageLimit差异、队列影响及代码示例咨询

一、ConcurrentMessageLimit的作用与影响 #

• Endpoint级别配置：控制整个接收端点同时处理的消息总数，是该端点下所有Consumer的并发处理上限。比如设为8，不管端点绑定了多少个Consumer，同一时间所有Consumer加起来最多只能处理8条消息。
• Consumer级别配置：针对单个Consumer的并发处理限制，优先级高于Endpoint配置。比如Endpoint设了10，某Consumer单独设为3，那这个Consumer最多同时处理3条消息，其余Consumer共享剩下的7个并发额度；如果多个Consumer的并发值总和超过Endpoint限制，最终会以Endpoint的上限为准。

简单来说，Endpoint是全局并发上限，Consumer是单个消费者的局部上限，局部限制不能突破全局限制。

二、同一队列消息分发给多个Consumer的代码示例 #

以.NET依赖注入环境下的RabbitMQ配置为例：

services.AddMassTransit(x =>
{
    // 注册需要绑定到同一队列的多个Consumer
    x.AddConsumer<OrderCreatedConsumer>();
    x.AddConsumer<OrderAuditConsumer>();
    x.AddConsumer<OrderNotificationConsumer>();

    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host("rabbitmq://localhost");

        // 配置接收端点，指定队列名称
        cfg.ReceiveEndpoint("order-events-queue", e =>
        {
            // 将多个Consumer关联到该端点
            e.ConfigureConsumer<OrderCreatedConsumer>(context);
            e.ConfigureConsumer<OrderAuditConsumer>(context);
            e.ConfigureConsumer<OrderNotificationConsumer>(context);

            // 可选：设置Endpoint级别的并发限制
            e.ConcurrentMessageLimit = 12;

            // 可选：给单个Consumer单独设置并发限制
            e.ConfigureConsumer<OrderNotificationConsumer>(context, c =>
            {
                c.ConcurrentMessageLimit = 3;
            });
        });
    });
});

配置完成后，order-events-queue队列中的每条消息都会被所有绑定在该端点的Consumer接收并处理。

内容的提问来源于stack exchange，提问作者David S.