问题描述
当通过Azure Event Hub SDK消费Event Hub中的消息时，必须指定一个Storage Account（存储账号）用于保存 Checkpoint (检查点)。
比如在C#代码中，需要指定Storage Account Connection String 和一个Blob Container的名称。其他语言代码也是一样，都需要指定Storage Account。
private const string ehubNamespaceConnectionString = "
private const string eventHubName = "
private const string blobStorageConnectionString = "
private const string blobContainerName = "
......
static async Task Main()
{
// Read from the default consumer group: $Default
string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;
// Create a blob container client that the event processor will use
storageClient = new BlobContainerClient(blobStorageConnectionString, blobContainerName);
// Create an event processor client to process events in the event hub
processor = new EventProcessorClient(storageClient, consumerGroup, ehubNamespaceConnectionString, eventHubName);
// Register handlers for processing events and handling errors
processor.ProcessEventAsync += ProcessEventHandler;
processor.ProcessErrorAsync += ProcessErrorHandler;
// Start the processing
await processor.StartProcessingAsync();
// Wait for 30 seconds for the events to be processed
await Task.Delay(TimeSpan.FromSeconds(30));
// Stop the processing
await processor.StopProcessingAsync();
}
那么，在Storage Account中如何查看Offest 和 Sequence Number的值呢？
解答如下
如上图，在Azure门户页面中，进入当前使用的Storage Account页面：
在指定的Container中，SDK会自动创建一个以Event Hub Namespace主机域名为名的Folder
然后是Event Hub -> 消费组（默认为 $default) --> Checkpint --> 分区号（从0开始，1，2 ...）
在以分区号为名称的Blob的元数据(Metadata)记录了sequencenumber 和 offset的值。
PS: 当删除或修改这个值后，如果重启Event Hub的消费端，这会导致数据从新的OFFSET开始获取。
参考资料
Checkpoint: https://docs.azure.cn/zh-cn/event-hubs/event-processor-balance-partition-load#checkpointing
向 Azure 事件中心发送事件及从 Azure 事件中心接收事件 - .NET (Azure.Messaging.EventHubs): https://docs.azure.cn/zh-cn/event-hubs/event-hubs-dotnet-standard-getstarted-send#receive-events
推荐阅读
发表评论