Azure 服务总线并非托管服务 Web 应用中收到的所有消息

Question

在 .net Web 应用程序中，我设置了一个托管服务来接收来自 Azure 服务总线主题的消息。问题是不是所有的消息都被接收到，只接收到任意数量的消息（例如，20 条消息只接收到 12 条）。其余的都进入了死信队列。当消息同时发送时会发生这种情况。 我尝试了以下步骤来解决这个问题：

1. 增加了最大并发调用量，这有帮助，但不能保证
2. 添加了预取计数

   我还尝试通过 Azure 服务总线资源中的功能发送消息。 500 条消息，没有间隔时间 --> 不起作用（对于所有消息）。 500条消息，1s间隔时间，所有消息都收到。

   我只是不明白为什么接收者没有收到所有的消息。 我想构建一个事件驱动的架构，如果所有消息都被处理，我不能让它成为一场赌博。

   启动.cs

   ...
   public void ConfigureServices(IServiceCollection services)
   {
       services.AddSingleton<IServiceBusTopicSubscription,ServiceBusSubscription>();
       services.AddHostedService<WorkerServiceBus>();
   }
   ...
           
   

   工人服务.cs

   public class WorkerServiceBus : IHostedService, IDisposable
   {
       private readonly ILogger<WorkerServiceBus> _logger;
       private readonly IServiceBusTopicSubscription _serviceBusTopicSubscription;
   
   
   
   
       public WorkerServiceBus(IServiceBusTopicSubscription serviceBusTopicSubscription,
           ILogger<WorkerServiceBus> logger)
       {
           _serviceBusTopicSubscription = serviceBusTopicSubscription;
           _logger = logger;
       }
   
   
   
      public async Task StartAsync(CancellationToken stoppingToken)
       {
           _logger.LogInformation("Starting the service bus queue consumer and the subscription");
           await _serviceBusTopicSubscription.PrepareFiltersAndHandleMessages().ConfigureAwait(false);
       }
   
   
   
      public async Task StopAsync(CancellationToken stoppingToken)
       {
           _logger.LogInformation("Stopping the service bus queue consumer and the subscription");
           await _serviceBusTopicSubscription.CloseSubscriptionAsync().ConfigureAwait(false);
       }
   
   
   
      public void Dispose()
       {
           Dispose(true);
           GC.SuppressFinalize(this);
       }
   
   
   
      protected virtual async void Dispose(bool disposing)
       {
           if (disposing)
           {
               await _serviceBusTopicSubscription.DisposeAsync().ConfigureAwait(false);
           }
       }
   }
   
   

   服务总线订阅.cs

   public class ServiceBusSubscription : IServiceBusTopicSubscription
   {
       private readonly IConfiguration _configuration;
       private const string TOPIC_PATH = "test";
       private const string SUBSCRIPTION_NAME = "test-subscriber";
       private readonly ILogger _logger;
       private readonly ServiceBusClient _client;
       private readonly IServiceScopeFactory _scopeFactory;
       private ServiceBusProcessor _processor;
   
   
   
      public ServiceBusBookingsSubscription(IConfiguration configuration,
           ILogger<ServiceBusBookingsSubscription> logger,
           IServiceScopeFactory scopeFactory)
       {
           _configuration = configuration;
           _logger = logger;
           _scopeFactory = scopeFactory;
   
   
   
          var connectionString = _configuration.GetConnectionString("ServiceBus");
           var serviceBusOptions = new ServiceBusClientOptions()
           {
               TransportType = ServiceBusTransportType.AmqpWebSockets
           };
           _client = new ServiceBusClient(connectionString, serviceBusOptions);
       }
   
   
   
      public async Task PrepareFiltersAndHandleMessages()
       {
           ServiceBusProcessorOptions _serviceBusProcessorOptions = new ServiceBusProcessorOptions
           {
               MaxConcurrentCalls = 200,
               AutoCompleteMessages = false,
               PrefetchCount = 1000,
           };
   
   
   
          _processor = _client.CreateProcessor(TOPIC_PATH, SUBSCRIPTION_NAME, _serviceBusProcessorOptions);
           _processor.ProcessMessageAsync += ProcessMessagesAsync;
           _processor.ProcessErrorAsync += ProcessErrorAsync;
   
   
   
          await _processor.StartProcessingAsync().ConfigureAwait(false);
       }
   
   
   
      private async Task ProcessMessagesAsync(ProcessMessageEventArgs args)
       {
           _logger.LogInformation($"Received message from service bus");
           _logger.LogInformation($"Message: {args.Message.Body}");
           var payload = args.Message.Body.ToObjectFromJson<List<SchedulerBookingViewModel>>();
   
   
   
          // Create scoped dbcontext
           using var scope = _scopeFactory.CreateScope();
           var dbContext = scope.ServiceProvider.GetRequiredService<dbContext>();
   
   
   
          // Process payload
           await new TestServiceBus().DoThings(payload);
   
   
   
          await args.CompleteMessageAsync(args.Message).ConfigureAwait(false);
       }
   
   
   
      private Task ProcessErrorAsync(ProcessErrorEventArgs arg)
       {
           _logger.LogError(arg.Exception, "Message handler encountered an exception");
           _logger.LogError($"- ErrorSource: {arg.ErrorSource}");
           _logger.LogError($"- Entity Path: {arg.EntityPath}");
           _logger.LogError($"- FullyQualifiedNamespace: {arg.FullyQualifiedNamespace}");
   
   
   
          return Task.CompletedTask;
       }
   
   
   
      public async ValueTask DisposeAsync()
       {
           if (_processor != null)
           {
               await _processor.DisposeAsync().ConfigureAwait(false);
           }
   
   
   
          if (_client != null)
           {
               await _client.DisposeAsync().ConfigureAwait(false);
           }
       }
   
   
   
      public async Task CloseSubscriptionAsync()
       {
           await _processor.CloseAsync().ConfigureAwait(false);
       }
   }
   

Answer 1

所以这就是我们解决问题的方法。它与为门户中的 Azure 资源设置的消息锁定持续时间有关。以前的值：30 秒。新值：3 分钟。