c# - 改进代码以根据上次消息日期从事件中心读取消息

标签 c# azure azureservicebus azure-iot-hub azure-eventhub

以下代码连接到 Azure 事件中心,它迭代所有分区,然后读取要处理的消息并将其插入数据库(待完成),代码工作正常,但是每次它都会读取所有消息。

这将作为 Azure WebJob 安装,因此它将连续、实时、不间断地运行。

  1. 如何改进此代码以仅读取未处理的消息?
  2. 是否有更好的方法来编码 while/for 部分,您会采用不同的方式吗?

    static void Main(string[] args)
    {
        ServiceBusConnectionStringBuilder builder = new ServiceBusConnectionStringBuilder(ConfigurationManager.AppSettings["ConnectionString"].ToString());
        builder.TransportType = TransportType.Amqp;
        MessagingFactory factory = MessagingFactory.CreateFromConnectionString(ConfigurationManager.AppSettings["ConnectionString"].ToString());
        EventHubClient client = factory.CreateEventHubClient(ConfigurationManager.AppSettings["eventHubEntity"].ToString());
        EventHubConsumerGroup group = client.GetDefaultConsumerGroup();
    
        CancellationTokenSource cts = new CancellationTokenSource();
        System.Console.CancelKeyPress += (s, e) =>
        {
            e.Cancel = true;
            cts.Cancel();
            Console.WriteLine("Exiting...");
        };
        var d2cPartitions = client.GetRuntimeInformation().PartitionIds;
    
        while (true)
        {
            foreach (string partition in d2cPartitions)
            {
                EventHubReceiver receiver = group.CreateReceiver(partition, DateTime.MinValue);
                EventData data = receiver.Receive();
                Console.WriteLine("{0} {1} {2}", data.PartitionKey, data.EnqueuedTimeUtc.ToLocalTime(), Encoding.UTF8.GetString(data.GetBytes()));
                var dateLastMessage = data.EnqueuedTimeUtc.ToLocalTime();
                receiver.Close();
                client.Close();
                factory.Close();
            }
        }
    }
    

最佳答案

使用 EventHubReceiver 无法为您提供所需的控制。相反,您应该使用 EventProcessorHost,它允许您使用检查点来恢复处理消息。

参见http://blogs.biztalk360.com/understanding-consumer-side-of-azure-event-hubs-checkpoint-initialoffset-eventprocessorhost/https://blogs.msdn.microsoft.com/servicebus/2015/01/16/event-processor-host-best-practices-part-1/用于背景阅读。

参见https://azure.microsoft.com/en-us/documentation/articles/event-hubs-csharp-ephcs-getstarted/#receive-messages-with-eventprocessorhost获取教程。

您可以轻松地在 WebJob 中托管 EventProcessor

关于c# - 改进代码以根据上次消息日期从事件中心读取消息,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/39084376/

相关文章:

c# - 使用 Java 构建应用程序与使用 C# 构建应用程序有什么不同吗?

c# - 使用 oauth 访问 gmail 的电子邮件

Azure Function 无法连接到 Cosmos DB

azure - 通过 Azure DocumentDB 中的 ID 访问资源

c# - Azure 服务总线中继安全

android - Azure 通知中心出现错误,并出现NotificationHubUnauthorizedException : Unauthorized in Android client

C# 检查 SQL 查询是否仅用于创建 View

c# - 使用 ServiceConfiguration (Azure) 中的连接字符串自定义 DbContext

json - 使用 Azure Function 向 azure 服务总线主题发送消息时,如何将内容类型指定为 application/json?

c# - 将文件从 Android 传输到 Windows PC 时指定的文件路径是什么?