MassTransit消费者无法对接Service Bus Topic问题排查

我正在构建一个.NET Core Worker应用，使用MassTransit消费Azure Service Bus Topic中的数据，但尝试多种方案后仍无法正常工作。

消费者类代码：

public class PublisherConsumer : IConsumer<NotificationPublishEvent>
{
    public Task Consume(ConsumeContext<NotificationPublishEvent> context)
    {
        var checkoutEvent = context.Message;
        // 处理消息逻辑
        Console.WriteLine($"Received Event with JobId: {checkoutEvent.JobId}");

        return Task.CompletedTask;
    }
}

Worker启动配置代码：

IHost host = Host.CreateDefaultBuilder(args)
    .ConfigureServices(services =>
    {
        
        var connectionString = "Endpoint=sb://xxx-notification.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=XXXZnHyPBx3YmqGNrKZa/0+XXX=";

        var subscriptionName = "notification-sub";

        services.AddMassTransit(config =>
        {
            config.SetKebabCaseEndpointNameFormatter();
            config.SetInMemorySagaRepositoryProvider();
            var entryAssembly = Assembly.GetEntryAssembly();
            config.AddConsumers(entryAssembly);
            config.AddSagaStateMachines(entryAssembly);
            config.AddSagas(entryAssembly);
            config.AddActivities(entryAssembly);
            config.UsingAzureServiceBus((context, cfg) =>
            {
                cfg.Host(connectionString);
                cfg.SubscriptionEndpoint<NotificationPublishEvent>(subscriptionName, ec =>
                {
                    ec.UseMessageRetry(x => x.Interval(5, TimeSpan.FromSeconds(1)));
                    ec.UseDelayedRedelivery(x => x.Incremental(5, TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(5)));
                    ec.ConfigureConsumer<PublisherConsumer>(context);
                });
            });
        });

        services.AddHostedService<Worker>();
    })
    .Build();

await host.RunAsync();

控制台日志显示总线已启动：Bus started: sb://aec-notification.servicebus.windows.net/，但订阅中已有18条消息，消费者却未触发。另外发现，我向notification主题发送消息且已创建notification-sub订阅，但日志显示自动生成了EventBus.Messages.Events相关主题及对应订阅，请问该现象原因及消费者不工作的问题根源是什么？

1. 自动生成EventBus.Messages.Events主题的原因 #

MassTransit默认会根据消息类型的命名空间+类型名，结合配置的命名格式化器（此处为KebabCase）生成主题名称。你的NotificationPublishEvent类型应该位于EventBus.Messages.Events命名空间下，所以MassTransit自动创建了对应的主题和订阅，而非使用你手动创建的notification主题。

2. 消费者不工作的根源 #

你手动创建的notification主题、notification-sub订阅，与MassTransit实际监听的目标不匹配：

• 你将消息发送到手动创建的notification主题，但MassTransit的消费者实际在监听自动生成的event-bus-messages-events-notification-publish-event主题的订阅。
• 两者主题不一致，导致notification-sub里的消息无法被当前消费者接收，自然触发不了消费逻辑。

方法一：指定MassTransit使用手动创建的主题 #

修改MassTransit配置，显式为NotificationPublishEvent指定对应主题名称为notification，让消费者监听你已创建的主题和订阅：

config.UsingAzureServiceBus((context, cfg) =>
{
    cfg.Host(connectionString);
    // 为消息类型指定自定义主题名
    cfg.Message<NotificationPublishEvent>(x =>
    {
        x.SetEntityName("notification"); // 匹配手动创建的主题名
    });

    cfg.SubscriptionEndpoint<NotificationPublishEvent>(subscriptionName, ec =>
    {
        ec.UseMessageRetry(x => x.Interval(5, TimeSpan.FromSeconds(1)));
        ec.UseDelayedRedelivery(x => x.Incremental(5, TimeSpan.FromMinutes(5), TimeSpan.FromMinutes(5)));
        ec.ConfigureConsumer<PublisherConsumer>(context);
    });
});

方法二：让MassTransit自动管理主题和订阅（推荐） #

如果不需要手动维护主题和订阅，可以删除已创建的notification主题和notification-sub订阅，让MassTransit自动生成对应资源。同时发送消息时使用MassTransit的API，确保消息发送到正确的自动生成主题：

var bus = serviceProvider.GetRequiredService<IBus>();
await bus.Publish<NotificationPublishEvent>(new
{
    JobId = "your-job-id" // 填充消息内容
});

额外检查项 #

• 确认NotificationPublishEvent的命名空间和类型名正确，避免因消息类型不匹配导致无法消费。
• 检查Azure Service Bus权限：确保连接字符串对应的Shared Access Key拥有Listen（订阅）和Send（发送消息）权限。
• 移除冗余Worker服务：如果仅用MassTransit消费者处理消息，services.AddHostedService<Worker>()可能多余，除非Worker类有其他业务逻辑。

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