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