MassTransit Consumer在Worker中无法消费SQS消息问题排查

我有一个服务向SQS的ta-emails队列发布消息，同时有一个运行中的Worker需要消费这些消息。已实现AdvisorRegisteredConsumer消费者类，并在Worker的Program.cs中配置了MassTransit对接Amazon SQS，appsettings.json配置了AWS相关参数，Worker基于BackgroundService实现。发布服务与消费Worker的MassTransit依赖注入配置完全相同，但消息能成功发布却从未被Worker中的消费者处理，怀疑是消费者或MassTransit配置遗漏了某些内容。

相关代码 #

消费者类 #

public class AdvisorRegisteredConsumer: IConsumer<AdvisorRegistered>
{
    private readonly ILogger<AdvisorRegisteredConsumer> _logger;

    public AdvisorRegisteredConsumer(ILogger<AdvisorRegisteredConsumer> logger)
    {
        _logger = logger;
    }

    public async Task Consume(ConsumeContext<AdvisorRegistered> context)
    {
        _logger.LogInformation("Consuming advisor id: {AdvisorId}", context.Message.AdvisorId);
    }
}

Worker的MassTransit配置 #

services.AddMassTransit(x =>
{
    x.UsingAmazonSqs((context, cfg) =>
    {
        AmazonSQSConfig _sqsConfig = new AmazonSQSConfig
        {
            ServiceURL = configuration.GetValue<string>("AWS:SQS:Endpoint"),
            AuthenticationRegion = configuration.GetValue<string>("AWS:Region")
        };

        cfg.Host(configuration.GetValue<string>("AWS:Region"), h =>
        {
            h.Config(_sqsConfig);
            h.AccessKey(configuration.GetValue<string>("AWS:SQS:AccessKey"));
            h.SecretKey(configuration.GetValue<string>("AWS:SQS:SecretAccessKey"));
        });
        
        cfg.ReceiveEndpoint("ta-emails", e =>
        {
            e.ConfigureConsumer<AdvisorRegisteredConsumer>(context);
        });
    });
});

AWS配置节点 #

"AWS": {
    "Region": "ru-central1",
    "SQS": {
        "Endpoint": "https://message-queue.api.cloud.yandex.net",
        "AccessKey": "REPLACED_MY_KEY_WITH_THIS",
        "SecretAccessKey": "REPLACED_MY_SECRET_KEY_WITH_THIS"
    }
},

Worker实现 #

public class Worker : BackgroundService
{
    private readonly ILogger<Worker> _logger;

    public Worker(ILogger<Worker> logger)
    {
        _logger = logger;
    }
    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
    {
        while (!cancellationToken.IsCancellationRequested)
        {
            _logger.LogInformation("Worker running at: {time}", DateTimeOffset.UtcNow);

            await Task.Delay(30000, cancellationToken);
        }
    }
}

排查与修复方案 #

1. 未注册消费者到MassTransit容器 #

当前配置仅在接收端点中调用ConfigureConsumer，但未将消费者注册到MassTransit的消费者集合中，导致MassTransit无法识别并实例化该消费者。

修复方法：在AddMassTransit配置中添加AddConsumer注册：

services.AddMassTransit(x =>
{
    // 新增：注册消费者到MassTransit容器
    x.AddConsumer<AdvisorRegisteredConsumer>();

    x.UsingAmazonSqs((context, cfg) =>
    {
        // 原有Host配置保持不变
        AmazonSQSConfig _sqsConfig = new AmazonSQSConfig
        {
            ServiceURL = configuration.GetValue<string>("AWS:SQS:Endpoint"),
            AuthenticationRegion = configuration.GetValue<string>("AWS:Region")
        };

        cfg.Host(configuration.GetValue<string>("AWS:Region"), h =>
        {
            h.Config(_sqsConfig);
            h.AccessKey(configuration.GetValue<string>("AWS:SQS:AccessKey"));
            h.SecretKey(configuration.GetValue<string>("AWS:SQS:SecretAccessKey"));
        });
        
        cfg.ReceiveEndpoint("ta-emails", e =>
        {
            e.ConfigureConsumer<AdvisorRegisteredConsumer>(context);
        });
    });
});

2. 自定义Worker的冗余性（可选） #

你的自定义Worker继承自BackgroundService，但MassTransit本身已经注册了独立的IHostedService来运行消费者逻辑。如果这个自定义Worker仅用于周期性日志输出，可考虑移除它，避免不必要的资源占用。

3. 验证消息类型一致性 #

确保发布端和消费端的AdvisorRegistered消息类完全一致（包括命名空间、属性名称与类型）。MassTransit依赖消息类型的完全限定名进行路由，两端定义不一致会导致消费者无法匹配消息。

4. 权限与队列配置检查 #

• 确认Worker使用的AWS密钥对ta-emails队列拥有接收消息的权限
• 验证Yandex Cloud SQS的端点配置正确，ru-central1区域的端点确实为https://message-queue.api.cloud.yandex.net
• 检查队列是否存在，且发布端确实向该队列发送消息

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