c# - MassTransit 未订阅 AzureServiceBus 主题

Question

我目前正在尝试使用 MassTransit 6.3.2 更新最初为 .NET Core 3.1 的应用程序。现在已配置为使用 .NET 6.0 和 MassTransit 7.3.0

我们的应用程序使用 MassTransit 通过 Azure 服务总线发送消息，将消息发布到主题，然后让其他订阅者监听这些主题。

精简一下，是这样实现的:

// Program.cs
services.AddMassTransit(config =>
{
   config.AddConsumer<AppointmentBookedMessageConsumer>();
   config.AddBus(BusControlFactory.ConfigureAzureServiceBus);
});


// BusControlFactory.cs
public static class BusControlFactory
{
   public static IBusControl ConfigureAzureServiceBus(IRegistrationContext<IServiceProvider> context)
   {
      var config = context.Container.GetService<AppConfiguration>();
      var azureServiceBus = Bus.Factory.CreateUsingAzureServiceBus(busFactoryConfig =>
      {
         busFactoryConfig.Host("Endpoint=sb://REDACTED-queues.servicebus.windows.net/;SharedAccessKeyName=MyMessageQueuing;SharedAccessKey=MyKeyGoesHere");
         busFactoryConfig.Message<AppointmentBookedMessage>(m => m.SetEntityName("appointment-booked"));
         busFactoryConfig.SubscriptionEndpoint<AppointmentBookedMessage>(
            "my-subscriber-name",
            configurator =>
            {
               configurator.UseMessageRetry(r => r.Interval(5, TimeSpan.FromSeconds(60)));
               configurator.Consumer<AppointmentBookedMessageConsumer>(context.Container);
            });
         return azureServiceBus;
      }
   }
}

现已更改并升级到最新的 MassTransit，其实现方式如下:

// Program.cs
services.AddMassTransit(config =>
{
   config.AddConsumer<AppointmentBookedMessageConsumer, AppointmentBookedMessageConsumerDefinition>();
   config.UsingAzureServiceBus((context, cfg) => 
   {
      cfg.Host("Endpoint=sb://REDACTED-queues.servicebus.windows.net/;SharedAccessKeyName=MyMessageQueuing;SharedAccessKey=MyKeyGoesHere");
      cfg.Message<AppointmentBookedMessage>(m => m.SetEntityName("appointment-booked"));

      cfg.ConfigureEndpoints(context);
});


// AppointmentBookedMessageConsumerDefinition.cs
public class AppointmentBookedMessageConsumerDefinition: ConsumerDefinition<AppointmentBookedMessageConsumer>
{
   public AppointmentBookedMessageConsumerDefinition()
   {
      EndpointName = "testharness.subscriber";
   }

   protected override void ConfigureConsumer(IReceiveEndpointConfigurator endpointConfigurator, IConsumerConfigurator<AppointmentBookedMessageConsumer> consumerConfigurator)
   {
      endpointConfigurator.UseMessageRetry(r => r.Interval(5, TimeSpan.FromSeconds(60)));
   }
}

问题(如果可以认为是一个问题的话)是我无法绑定(bind)到已存在的订阅。

在上面的示例中，您可以看到 EndpointName 设置为“testharness.subscriber”。在我升级之前，已经有一个主题“预约”的订阅。但是，当应用程序运行时，它不会出错，但不会收到任何消息。

如果我将 EndpointName 更改为“testharness.subscriber2”。另一个订阅者出现在 Azure 服务总线主题中(通过 Azure 门户)，我开始接收消息。我看不出名称有什么区别(除了我所做的更改，在本例中:“2”后缀)。

我在这里遗漏了什么吗？我还需要做些什么来绑定(bind)它们吗？是我的配置错误吗？是不是错了？虽然我确信我可以通过更密切地管理版本并在使用新队列时删除不需要的队列来解决这个问题，但感觉这是错误的方法。

Answer 1

使用 Azure 服务总线，订阅上的 ForwardTo 可能有点不透明。

虽然订阅确实可能在视觉上表明它正在转发到正确命名的队列，但可能队列已被删除并在某个时刻重新创建，而没有删除订阅。这会导致订阅将构建消息，因为它无法将消息转发到不再存在的队列。

为什么？在内部，订阅将 ForwardTo 维护为 对象 ID，在删除队列后，该对象 ID 指向不存在的对象 - 导致消息在订阅中堆积.

如果您的订阅中有消息，您可能需要进入门户并更新该订阅以指向新队列(即使它具有相同的名称)，此时消息应该流向队列。

如果订阅中没有任何消息(或者它们不重要)，您只需删除订阅即可，当您重新启动公交车时，MassTransit 将会重新创建该订阅。