如何将EventProcessorHost设置为从现在起读取事件(UTC)

Question

我们正在使用EventProcessorHost从Azure EventHubs接收事件。我一直试图（通过EventProcessorOptions.InitialOffsetProvider）将其配置为从UTC读取事件，但从现在开始，它总是从提要的开头读取。我没有保存检查点（我甚至删除了创建的BLOB容器）。这就是我设置它的方式：

DateTime startDate = DateTime.UtcNow;
var epo = new EventProcessorOptions
            {
                MaxBatchSize = 100, 
                PrefetchCount = 100, 
                ReceiveTimeOut = TimeSpan.FromSeconds(120),  
                InitialOffsetProvider = (name) => startDate  
            };

如有任何指导，我们将不胜感激。

Answer 1

认为这在2.0.0版本中发生了变化-Rajiv的代码现在是：

var eventProcessorOptions = new EventProcessorOptions
{
    InitialOffsetProvider = (partitionId) => EventPosition.FromEnqueuedTime(DateTime.UtcNow)
};

下面是一个具有完全限定类名的示例块：

    private static async Task MainAsync(string[] args)
    {
        try{
            Console.WriteLine("Registering EventProcessor...");
            string AISEhConnectionStringEndpoint = Configuration["AISEhConnectionStringEndpoint"];
            string AISEhConnectionStringSharedAccessKeyName = Configuration["AISEhConnectionStringSharedAccessKeyName"];
            string AISEhConnectionStringSharedAccessKey = Configuration["AISEhConnectionStringSharedAccessKey"];
            string EhConnectionString = $"Endpoint={AISEhConnectionStringEndpoint};SharedAccessKeyName={AISEhConnectionStringSharedAccessKeyName};SharedAccessKey={AISEhConnectionStringSharedAccessKey}";
            string AISEhEntityPath = Configuration["AISEhEntityPath"];
            string AISEhConsumerGroupName = Configuration["AISEhConsumerGroupName"];
            string AISStorageContainerName = Configuration["AISStorageContainerName"];
            string AISStorageAccountName = Configuration["AISStorageAccountName"];
            string AISStorageAccountKey = Configuration["AISStorageAccountKey"];
            string StorageConnectionString = string.Format("DefaultEndpointsProtocol=https;AccountName={0};AccountKey={1}", AISStorageAccountName, AISStorageAccountKey);
            var eventProcessorHost = new Microsoft.Azure.EventHubs.Processor.EventProcessorHost(
                AISEhEntityPath,
                AISEhConsumerGroupName,
                EhConnectionString,
                StorageConnectionString,
                AISStorageContainerName);
            var options = new Microsoft.Azure.EventHubs.Processor.EventProcessorOptions
            {
                InitialOffsetProvider = (partitionId) => Microsoft.Azure.EventHubs.EventPosition.FromEnqueuedTime(DateTime.UtcNow)
            };
            // Registers the Event Processor Host and starts receiving messages
            await eventProcessorHost.RegisterEventProcessorAsync<GetEvents>(options);
            Thread.Sleep(Timeout.Infinite);
            // Disposes of the Event Processor Host
            await eventProcessorHost.UnregisterEventProcessorAsync();
        }
        catch(Exception ex)
        {
            Console.WriteLine(ex.Message);
            NLog.LogManager.GetCurrentClassLogger().Error(ex);
            throw;
        }
    }
}

以下是我的一般设置，其中隐藏了秘密/确切地址，以帮助解决问题，因为我发现解决这个问题比拔牙更不愉快：

"AISEhConnectionStringEndpoint": "sb://<my bus address>.servicebus.windows.net/",
"AISEhConnectionStringSharedAccessKeyName": "<my key name>",
"AISEhConnectionStringSharedAccessKey": "<yeah nah>",
"AISEhEntityPath": "<Event Hub entity path>",
"AISEhConsumerGroupName":  "<consumer group name e.g $Default>",
"AISStorageContainerName":  "<storage container name>",
"AISStorageAccountName": "<storage account name>",
"AISStorageAccountKey": "<yeah nah>",

Answer 2

我发现blob中的检查点文件夹仍然存在，我的应用程序正在考虑这一点，并忽略了我在EventProcessorOptions中设置的日期。在我删除容器后，它开始按预期运行（计入UTC日期）。

Answer 3

您可以为此使用EventProcessorOptions类，并提供所需时间的偏移集。

var eventProcessorOptions = new EventProcessorOptions
{
    InitialOffsetProvider = (partitionId) => DateTime.UtcNow
};

然后，您可以使用任何接受eventProcessorOptions的RegisterEventProcessAsync重载。