Получение данных из EventHub задерживается

Question

У меня есть EventHub, настроенный в Azure, а также группа потребителей для чтения данных. Он работал нормально в течение нескольких дней. Внезапно я вижу, что задержка входящих данных (около 3 дней). Я использую Windows Service для использования данных на моем сервере. У меня около 500 входящих сообщений в минуту. Может ли кто-нибудь помочь мне понять это?

Answer 1

Возможно, вы слишком медленно обрабатываете их. Поэтому работа над этим будет расти, и вы отстанете.

Чтобы получить некоторое представление в том, где вы находитесь в случае поток можно использовать такой код:

private void LogProgressRecord(PartitionContext context) 
{ 
    if (namespaceManager == null) 
     return; 

    var currentSeqNo = context.Lease.SequenceNumber; 
    var lastSeqNo = namespaceManager.GetEventHubPartition(context.EventHubPath, context.ConsumerGroupName, context.Lease.PartitionId).EndSequenceNumber; 
    var delta = lastSeqNo - currentSeqNo; 

    logWriter.Write(
      $"Last processed seqnr for partition {context.Lease.PartitionId}: {currentSeqNo} of {lastSeqNo} in consumergroup '{context.ConsumerGroupName}' (lag: {delta})", 
      EventLevel.Informational); 
} 

namespaceManager строится так:

namespaceManager = NamespaceManager.CreateFromConnectionString("Endpoint=sb://xxx.servicebus.windows.net/;SharedAccessKeyName=yyy;SharedAccessKey=zzz"); 

Я называю этот метод протоколирования в CloseAsync метод:

public Task CloseAsync(PartitionContext context, CloseReason reason) 
{ 
    LogProgressRecord(context); 

    return Task.CompletedTask; 
} 

logWriter лишь некоторые Loggi ng класс, который я использовал для записи информации в хранилище blob.

Теперь она выводит сообщения, как

Последняя переработанного seqnr для раздела 3: 32780931 из 32823804 в consumergroup 'телеметрической' (лаг: 42873)

так, когда отставание очень высок, вы могли бы обрабатывать события, которые произошли давно. В этом случае вам нужно увеличить/увеличить процессор.

Если вы заметили задержку, вы должны измерить, сколько времени требуется для обработки заданного количества предметов. Затем вы можете попытаться оптимизировать производительность и посмотреть, улучшится ли она. Мы сделали это как:

public async Task ProcessEventsAsync(PartitionContext context, IEnumerable<EventData> events) 
{ 
     try 
     { 
      stopwatch.Restart(); 

      // process items here 

      stopwatch.Stop(); 

      await CheckPointAsync(context); 

      logWriter.Write(
       $"Processed {events.Count()} events in {stopwatch.ElapsedMilliseconds}ms using partition {context.Lease.PartitionId} in consumergroup {context.ConsumerGroupName}.", 
       EventLevel.Informational); 
     } 
}