如何配置MassTransit实现每条消息仅被单个消费者实例接收

问题分析 #

你期望实现的是同类型消息的竞争消费模式：单条消息仅被多个消费者中的一个处理一次，但你当前的代码配置违背了MassTransit的投递规则：
同一个接收端点（ReceiveEndpoint）下注册多个消费同一消息类型的不同消费者时，MassTransit会将每条消息广播给所有匹配的消费者，会导致同一条消息被多次处理，完全不符合需求。

正确实现方案 #

根据你的实际业务场景，可选择以下两种方案：

方案一：消费者逻辑一致，仅配置不同（推荐） #

你的两个消费者仅泛型参数不同、逻辑完全一致的情况下，无需定义多个泛型消费者实例，直接将配置动态注入单个消费者类即可，修改后参考代码如下：

// 只注册一个通用消费者，配置通过IOption或者构造函数注入
services.AddScoped<S3ToArchiveConsumer>();     
services.AddMassTransit(x =>
{
    x.SetKebabCaseEndpointNameFormatter();
    x.UsingRabbitMq((context, cfg) =>
    {
        CaringoShovelSettings caringoShovel = context.GetRequiredService<IOptions<CaringoShovelSettings>>().Value;
        AppSettings appSettings = context.GetRequiredService<IOptions<AppSettings>>().Value;
        cfg.Host(appSettings.RabbitMqHost, "/", h =>
        {
            h.Username(appSettings.RabbitMqUsername);
            h.Password(appSettings.RabbitMqPassword);
        });
        if ((caringoShovel.WasabiEastOneShovel?.Enabled ?? false) || (caringoShovel.WasabiEastTwoShovel?.Enabled ?? false))
        {
            cfg.ReceiveEndpoint(nameof(MoveCaringoFileRequest), e =>
            {
                e.Durable = true;
                e.ThrowOnSkippedMessages();
                // 统一注册单个消费者，并发数取所有启用配置的总和，内部动态选择对应配置执行
                e.Consumer<S3ToArchiveConsumer>(context, configure =>
                {
                    var totalConcurrency = 0;
                    var maxTimeout = TimeSpan.FromSeconds(30);
                    if (caringoShovel.WasabiEastOneShovel?.Enabled ?? false)
                    {
                        totalConcurrency += caringoShovel.WasabiEastOneShovel.Concurrency;
                        maxTimeout = caringoShovel.WasabiEastOneShovel.Timeout > maxTimeout ? caringoShovel.WasabiEastOneShovel.Timeout : maxTimeout;
                    }
                    if (caringoShovel.WasabiEastTwoShovel?.Enabled ?? false)
                    {
                        totalConcurrency += caringoShovel.WasabiEastTwoShovel.Concurrency;
                        maxTimeout = caringoShovel.WasabiEastTwoShovel.Timeout > maxTimeout ? caringoShovel.WasabiEastTwoShovel.Timeout : maxTimeout;
                    }
                    configure.UseTimeout(x => x.Timeout = maxTimeout);
                    configure.UseConcurrentMessageLimit(totalConcurrency);
                });
            });
        }
        cfg.ConfigureEndpoints(context);
    });
});
services.AddMassTransitHostedService();

这种配置下，同一条消息只会被一个消费者实例处理，自动进入竞争消费模式，可通过调整并发数控制处理速度。

方案二：消费者逻辑完全不同，需要按规则选择执行 #

如果两个消费者的处理逻辑差异很大，你可以新增一个统一的入口消费者接收所有消息，在内部根据你的规则（比如配置优先级、消息属性）分发到对应消费者执行，确保同一条消息仅触发一次处理逻辑。

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