MassTransit不订阅AzureServiceBus主题

Question

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

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

削减后，它的实现方式如下：

// 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)));
   }
}

如果它可以被认为是一个问题，那就是我无法绑定到已经存在的订阅。

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

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

我是不是漏掉了什么？我还需要做点什么才能把这些绑起来吗？我的配置有问题吗？是错的吗？虽然我确信我可以通过更紧密地管理发行版并在使用新队列之后删除不需要的队列来解决这一问题，但这似乎是错误的方法。

Answer 1

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

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

为什么？在内部，订阅将ForwardTo维护为对象id，在删除队列后，该id指向订阅中生成的消息中不存在- resulting的对象。

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

如果订阅中没有任何消息(或者它们不重要)，您可以只删除订阅，在重新启动总线时，它将由MassTransit重新创建。