两个MassTransit消费者同时消费同一请求的问题排查

你遇到的核心问题是同一个请求消息被多个Worker实例的消费者并发处理，导致重复调用第三方API触发409冲突，根源在于MassTransit的端点配置逻辑错误：

1. 手动创建ReceiveEndpoint与ConsumerDefinition配置冲突
   你已经在RequestConsumer_MyConsumer_Definition中指定了端点名称myconsumer，但在UsingRabbitMq配置块里又手动创建了一个未命名的ReceiveEndpoint并绑定消费者。这会导致MassTransit忽略ConsumerDefinition的端点配置，生成默认队列绑定，最终两个Worker实例都监听同一个队列，且由于配置异常触发了广播式消费（而非正常的竞争消费）。

2. 请求消息处理模式未匹配
   针对IRequest_MyRequest这类请求型消息，需要依赖MassTransit的请求-响应机制确保单实例消费，但你的配置没有正确利用该机制，导致消息被多实例同时拾取处理。

1. 修正MassTransit服务配置 #

移除手动创建的ReceiveEndpoint，改用ConfigureEndpoints让MassTransit自动根据ConsumerDefinition生成端点，这样定义中的队列名称、并发限制、重试策略等配置会完全生效：

public static void ConfigureWorkerServices(IServiceCollection services, IConfiguration configuration){

services.AddOptions<MassTransitHostOptions>()
    .Configure(options =>
    {
        options.WaitUntilStarted = true;
        options.StartTimeout = TimeSpan.FromSeconds(10);
        options.StopTimeout = TimeSpan.FromSeconds(30);
    });

services.AddMassTransit(x =>
{        
    x.AddConsumer<RequestConsumer_MyConsumer, RequestConsumer_MyConsumer_Definition>();
    
    x.UsingRabbitMq((context, cfg) =>
    {
        var rabbitMqUsername = configuration.GetValue<string>("RabbitMqUsername");
        var rabbitMqPassword = configuration.GetValue<string>("RabbitMqPassword");
        var rabbitMqHost = configuration.GetValue<string>("RabbitMqHost");
        var rabbitMqPort = configuration.GetValue<int>("RabbitMqPort");
        
        cfg.Durable = false;
        cfg.AutoDelete = true;
        cfg.SetQueueArgument("x-expires", null);
        cfg.Host(rabbitMqHost, rabbitMqPort, "/", h =>
            {
                h.Username(rabbitMqUsername);
                h.Password(rabbitMqPassword);
                h.UseSsl(s =>
                {
                    s.AllowPolicyErrors(SslPolicyErrors.RemoteCertificateNameMismatch);
                });
            });            
        
        // 替换手动端点配置，自动应用ConsumerDefinition中的设置
        cfg.ConfigureEndpoints(context);
    });
});

2. 确保请求消息使用Send而非Publish #

发送IRequest_MyRequest时，必须用Send方法将消息定向到myconsumer队列，不能用Publish（Publish会广播消息到所有订阅队列）：

// 示例：正确的请求发送方式
var endpoint = await bus.GetSendEndpoint(new Uri($"rabbitmq://{rabbitMqHost}/myconsumer"));
await endpoint.Send<IRequest_MyRequest>(new { /* 请求参数 */ });

3. 验证RabbitMQ队列状态 #

修复配置后，登录RabbitMQ控制台确认：

• 存在唯一的myconsumer队列
• 两个Worker实例作为该队列的消费者（处于竞争消费模式，同一消息只会被一个实例拾取）
• 队列未绑定到fanout类型交换器

• ConsumerDefinition中的ConcurrentMessageLimit是单实例的并发数，整个队列的总并发数为实例数×单实例并发数，可根据业务需求调整
• 已配置的重试策略会作用于单个消费者实例，避免因重试导致的重复调用第三方API

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