基于RabbitMQ-MassTransit的消息并发处理方案咨询

核心思路 #

• 定义固定数量的分区队列（比如8个，可根据并发需求调整），通过一致性哈希将同一RequestId的消息路由到同一个队列
• 每个分区队列的消费者设置为串行处理，确保同一RequestId的消息按顺序执行
• 支持水平扩展：部署多个消费者实例时，RabbitMQ会自动将队列的消费负载分配到不同实例

步骤1：生产者（网关）配置 #

发送消息时指定分区键为RequestId，让MassTransit/RabbitMQ自动路由到对应分区队列：

// 网关项目中的MassTransit配置
services.AddMassTransit(config =>
{
    config.SetKebabCaseEndpointNameFormatter();

    config.UsingRabbitMq((context, cfg) =>
    {
        var settings = context.GetRequiredService<IOptions<RabbitMQSettings>>().Value;
        cfg.Host(settings.Host, "/", h =>
        {
            h.Username(settings.Username);
            h.Password(settings.Password);
        });

        // 配置消息发布使用一致性哈希交换器
        cfg.Publish<RequestMessage>(p =>
        {
            p.SetExchangeType("x-modulus-hash");
            p.Durable = true;
        });
    });
});

// 发送消息的业务代码示例
public async Task SendRequest(RequestMessage message)
{
    var endpoint = await _bus.GetPublishSendEndpoint<RequestMessage>();
    await endpoint.Send(message, context =>
    {
        // 设置分区键为RequestId，确保同一RequestId路由到同一队列
        context.SetPartitionKey(message.RequestId.ToString());
    });
}

步骤2：消费者Web API配置 #

推荐使用MassTransit内置的UsePartition特性自动配置分区队列，无需手动创建多个端点：

services.AddMassTransit(config =>
{
    config.AddConsumer<MessageRequestConsumer>();

    config.UsingRabbitMq((context, cfg) =>
    {
        var settings = context.GetRequiredService<IOptions<RabbitMQSettings>>().Value;
        cfg.Host(settings.Host, "/", h =>
        {
            h.Username(settings.Username);
            h.Password(settings.Password);
        });

        cfg.ReceiveEndpoint(settings.QueueName, e =>
        {
            // 启用分区：指定8个分区，以RequestId作为分区键
            e.UsePartition(8, context =>
            {
                var message = context.Message as RequestMessage;
                return message?.RequestId.ToString() ?? string.Empty;
            });

            // 关键配置：每个分区队列仅允许1个并发消费，确保串行处理
            e.ConcurrentMessageLimit = 1;
            // 单次预取1条消息，避免跨实例抢占同一RequestId的消息
            e.PrefetchCount = 1;

            e.ConfigureConsumer<MessageRequestConsumer>(context);
        });

        cfg.UseConsumeFilter(typeof(ConsumerLoggingMiddleware<>), context);
    });
});

手动配置分区队列（可选） #

如果需要更精细的控制，可手动创建分区队列并绑定到哈希交换器：

services.AddMassTransit(config =>
{
    config.AddConsumer<MessageRequestConsumer>();

    config.UsingRabbitMq((context, cfg) =>
    {
        var settings = context.GetRequiredService<IOptions<RabbitMQSettings>>().Value;
        cfg.Host(settings.Host, "/", h =>
        {
            h.Username(settings.Username);
            h.Password(settings.Password);
        });

        int partitionCount = 8;
        var exchangeName = typeof(RequestMessage).Name;

        for (int i = 0; i < partitionCount; i++)
        {
            var queueName = $"{settings.QueueName}-partition-{i}";
            cfg.ReceiveEndpoint(queueName, e =>
            {
                e.ConfigureConsumeTopology = false;
                e.ConcurrentMessageLimit = 1;
                e.PrefetchCount = 1;

                // 绑定到一致性哈希交换器，指定路由键为分区索引
                e.Bind(exchangeName, b =>
                {
                    b.ExchangeType = "x-modulus-hash";
                    b.RoutingKey = i.ToString();
                    b.Arguments = new Dictionary<string, object>
                    {
                        { "hash-header", "partition-key" }
                    };
                });

                e.ConfigureConsumer<MessageRequestConsumer>(context);
            });
        }

        cfg.UseConsumeFilter(typeof(ConsumerLoggingMiddleware<>), context);
    });
});

关键配置说明 #

1. 分区数量：建议设置为CPU核心数的2-4倍（如8、16），平衡并发能力与资源占用
2. ConcurrentMessageLimit=1：强制每个分区队列的消息串行处理，同一RequestId的消息进入同一队列后会按顺序执行
3. 一致性哈希逻辑：RabbitMQ的x-modulus-hash交换器会根据partition-key头计算哈希值，取模后路由到对应分区队列，确保同一RequestId的消息始终进入同一队列
4. 水平扩展：部署多个消费者实例时，RabbitMQ会自动将不同分区队列的消费任务分配到各个实例，实现负载均衡

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