從EventHub獲取數據被延遲

Question

我有一個在Azure中配置的EventHub，也是一個用於讀取數據的使用者組。它在一些日子裏工作正常。突然間，我看到傳入數據有所延遲（大約3天）。我使用Windows服務來使用服務器中的數據。每分鐘有大約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的在consumergroup '遙測' 32823804（滯後：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); 
     } 
}