c# - 如何在 C# 中为事件中心创建接收器以获取给定时间跨度的消息？

Question

using Azure.Messaging.EventHubs.Consumer;
using System.Text;

string connectionString = "";
string eventHubName = "";
string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;

await using (var consumerClient = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName))
{
  await foreach (PartitionEvent partitionEvent in consumerClient.ReadEventsAsync())
  {
    string json = Encoding.UTF8.GetString(partitionEvent.Data.Body.ToArray());
    Console.WriteLine(json);
  }
}

当前返回来自 eventHub 的所有消息。我只需要接收过去 1 小时的消息，然后停止收听新消息。

Answer 1

谢谢@FalcoAlexander，指出了正确的方向。我已经提到了Read events from a partition, starting from a specific date and time并且能够对您的代码进行以下更改，这给了我预期的结果。

• 我没有使用 ReadEventsAsync()，而是使用了 ReadEventsFromPartitionAsync，我可以在其中提及我想要的时间段。
• 为了重现相同的内容，我认为时间为 2 分钟，但根据您的情况，您可以需要 1 小时。

using Azure.Messaging.EventHubs.Consumer;
using System.Diagnostics;
using System.Text;

string connectionString = "";
string eventHubName = "";
string consumerGroup = EventHubConsumerClient.DefaultConsumerGroupName;


await using (var consumerClient = new EventHubConsumerClient(consumerGroup, connectionString, eventHubName))
{
    DateTimeOffset timesamp = DateTimeOffset.UtcNow.Subtract(TimeSpan.FromMinutes(10)); // In your case this line would be DateTimeOffset timesamp = DateTimeOffset.UtcNow.Subtract(TimeSpan.FromHours(1));
    EventPosition timestamp = EventPosition.FromEnqueuedTime(timesamp);
    using CancellationTokenSource cancellationSource = new CancellationTokenSource();
    string partitionId = (await consumerClient.GetPartitionIdsAsync(cancellationSource.Token)).First();

    await foreach (PartitionEvent partitionEvent in consumerClient.ReadEventsFromPartitionAsync(
       partitionId,
        timestamp))
    {
        string json = Encoding.UTF8.GetString(partitionEvent.Data.Body.ToArray());
        Console.WriteLine(json);
    }
}

结果: