使用MassTransit+RabbitMQ发布消息仅随机触发一个消费者如何解决

问题根因 #

你两个独立的消费者服务都绑定了同一个队列PatintQueue，RabbitMQ默认对同一个队列的多个消费者采用竞争消费模式，每条消息只会随机投递给其中一个消费者，不会重复投递，因此会出现每次仅触发一个消费者的现象。

解决方法 #

要实现单条消息被多个独立消费者各消费一次，需要让每个消费者服务绑定自己独有的队列，匹配MassTransit的发布/订阅模式，按以下规则修改消费者端配置即可：

1. 删掉固定队列名的硬编码声明，优先使用MassTransit自动生成端点的能力，或者为每个消费者服务指定唯一的队列名
2. 确保消费者和队列正确绑定

推荐配置（自动生成独立队列） #

#region Service Bus 
public static IServiceCollection AddEventBus(this IServiceCollection services, IConfiguration configuration)
{
    services.AddMassTransit(x =>
    {
        RegisterMessageConsumers(x);
        RegisterRequestClients(x);
        x.UsingRabbitMq((context, cfg) =>
        {
            cfg.Host(configuration["EventBusConnection"], h =>
            {
                h.Username(configuration["EventBusUserName"]);
                h.Password(configuration["EventBusPassword"]);
            });
            // 自动按消费者类型生成唯一队列，自动绑定消息交换机
            cfg.ConfigureEndpoints(context);
        });
    });
    services.AddMassTransitHostedService();
    return services;
}

private static void RegisterMessageConsumers(IServiceCollectionBusConfigurator service)
{
    service.AddConsumer<ConsultantFinishedEventConsumer>();
}
#endregion

手动指定队列名的配置 #

如果需要自定义队列名，要确保不同消费者服务的队列名唯一，同时手动绑定消费者：

// 消费者服务A配置
cfg.ReceiveEndpoint("ConsultantFinishedEvent_ServiceA_Queue", ep =>
{
    ep.PrefetchCount = 16;
    ep.ConfigureConsumer<ConsultantFinishedEventConsumer>(context);
});

// 消费者服务B配置
cfg.ReceiveEndpoint("ConsultantFinishedEvent_ServiceB_Queue", ep =>
{
    ep.PrefetchCount = 16;
    ep.ConfigureConsumer<ConsultantFinishedEventConsumer>(context);
});

额外注意点 #

• 如果多个消费者服务里的消费者类全类名完全一致，自动生成的队列名也会重复，此时可以给消费者类加EndpointName特性指定唯一队列名：

[EndpointName("ConsultantFinishedEvent_ServiceA_Queue")]
public class ConsultantFinishedEventConsumer : IConsumer<ConsultantFinishedEvent>
{
    public Task Consume(ConsumeContext<ConsultantFinishedEvent> context)
    {
        // 业务逻辑
        return Task.CompletedTask;
    }
}

• 原消费者代码中Consume方法返回null会引发异常，需要替换为return Task.CompletedTask。

内容的提问来源于stack exchange，提问作者穆罕默德 - Moh