【Azure 事件中心】Event Hubs如何获取其中存放的历史消息

Hubs,Event ,历史,消息 · 浏览次数 : 7

小编点评

**获取 Event Hub 中存储的历史消息的步骤:** 1. **确认消息进入 Event Hub 的保留期限内。** - 使用 `EventPosition.RetentionDays` 属性获取保留期限。 - 如果消息超过保留期限,它将被删除。 2. **指定获取消息的开始时间或偏移量。** - 使用 `fromEnqueuedTime` 或 `EventPosition.Earliest` 方法指定开始时间或事件位置。 3. **使用 EventHubClient 或 EventHubReceivedMessage 接口接收事件。** - 接收事件时,可根据时间或事件位置检索相应的消息。 4. **处理接收到的消息。** - 处理接收到的消息并进行相应的处理。 **代码示例:** ```java // 使用 EventHubClient 获取 Event Hub 的实例 EventHubClient eventHubClient = EventHubClient.builder() .connectionString(connectionString) .build(); // 获取事件的开始时间 long startTime = eventHubClient.getUtcTime().toEpochMilli(); // 使用 EventPosition.Earliest 获取最早的消息 EventPosition earliestPosition = eventHubClient.positionFromEnqueuedTime(startTime); // 接收事件 while (true) { EventHubReceivedMessage message = eventHubClient.receiveEventHubMessages(10); for (EventHubReceivedMessage receivedMessage : message) { // 处理接收到的消息 } } ``` **注意:** - 在使用 `fromEnqueuedTime` 或 `EventPosition.Earliest` 方法获取消息之前,请确保事件已进入 Event Hub 的保留期限内。 - 您可以根据需要调整 `startTime` 的值来获取更早的事件。

正文

问题描述

使用Azure Event Hub服务,除了正常的生产,消费消息以外,如果想拿到Event Hub中存储的历史消息?有什么方法呢?

 

问题解答

获取 Event Hubs 存储的历史消息,首先需要确保消息进入Event Hub的时间处于保留期限(Retention Days)内,因为超过这个期限消息会被删除。

如果确认在期限内,在代码中可以指定时间或者对应的offset获取对应的消息。以Java为例可以参考以下sample:

如果需要从最早的消息,可以直接使用 EventPosition.Earliest(),也可以使用fromEnqueuedTime,参照:https://github.com/Azure/azure-sdk-for-java/blob/main/sdk/eventhubs/azure-messaging-eventhubs/src/main/java/com/azure/messaging/eventhubs/models/EventPosition.java

当然,使用Service Bus Explorer工具也能获取到Event Hub中的数据,下载地址:https://github.com/paolosalvatori/ServiceBusExplorer

当创建Listener 之后,可以指定start time

 

参考资料

Event Hub事件保留:https://docs.azure.cn/zh-cn/event-hubs/event-hubs-features#event-retention

 

 

与【Azure 事件中心】Event Hubs如何获取其中存放的历史消息相似的内容:

【Azure 事件中心】Event Hubs如何获取其中存放的历史消息

问题描述 使用Azure Event Hub服务,除了正常的生产,消费消息以外,如果想拿到Event Hub中存储的历史消息?有什么方法呢? 问题解答 获取 Event Hubs 存储的历史消息,首先需要确保消息进入Event Hub的时间处于保留期限(Retention Days)内,因为超过这个

【Azure 事件中心】Kafka 生产者发送消息失败,根据失败消息询问机器人得到的分析步骤

问题描述 Azure Event Hubs -- Kafka 生产者发送消息存在延迟接收和丢失问题, 在客户端的日志中发现如下异常: 2023-06-05 02:00:20.467 [kafka-producer-thread | producer-1] ERROR com.deloitte.com

【Azure 事件中心】Event Hubs中存在非常多的错误数据,是否能提前删除这些数据呢?

问题描述 因为一些特殊原因,Event Hub 里面堆积了很多不需要的数据事件,正常要等事件中的过期时间到后才有Event Hub自动删除掉,但希望能够尽快马上删除,有没有什么手动的方法吗? 问题解答 Event Hub是一个数据事件处理服务,最主要的功能就是:接收和发送事件。它并不是一个数据存储服

【Azure 事件中心】使用Apache Flink 连接 Event Hubs 出错 Kafka error: No resolvable bootstrap urls

问题描述 参考Github上 Event Hub的示例代码(Using Apache Flink with Event Hubs for Apache Kafka Ecosystems : https://github.com/Azure/azure-event-hubs-for-kafka/tre

【Azure 事件中心】Spring Cloud Stream Event Hubs Binder 发送Event Hub消息遇见 Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially 异常

Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially

【Azure 事件中心】Event Hub 无法连接,出现 Did not observe any item or terminal signal within 60000ms in 'flatMapMany' 的错误消息

2022-11-03 10:58:21.474 INFO --- [pool-7-thread-1] c.a.m.e.PartitionBasedLoadBalancer []: Load balancer already running 2022-11-03 10:58:51.014 WARN --- [ parallel-2] c.a.m.e.Partition

【Azure 事件中心】Event Hub服务中的度量值指标介绍

问题描述 Event Hub服务中的度量值指标解说 1)request和message的区别 2)capture backlog 和 capture message 怎么理解 3)quota exceed error, 哪里可以知道这个quota 4)size 是指的存储大小 5) 为EventHu

【Azure 事件中心】Azure Event Hub中的数据能不能存储大于7天呢?如果7天之后是不是会自动删除呢?

问题描述 Event Hub中有个retention的设置为7天,有没有办法增大这个Retention的时间? 如果没办法,是不是超过7天的数据就会被删除? 问题解答 因为Azure Event Hub(事件中心)是一个实时事件流引擎,其设计意图并不是用于代替数据库以及/或者用作无限期保存的事件流的

【Azure 事件中心】Azure Event Hub客户端遇见 Expired Heartbeat 错误

问题描述 Azure Event Hub 在消费数端中,经常性遇见 Expired Heartbeat 错误 (consumer-xxxxxxxxxxxxx-c84873c6c828e8df6c843861ad36affb from group xxxxxxxxxxxx due to expired

【Azure 事件中心】向Event Hub发送数据异常 : partitionId[null]: Sending messages timed out

问题描述 在使用Java 代码向 Azure Event Hub发送数据时,先后遇见了如下两种异常消息: 1)ERROR c.t.d.h.s.source.EventHubLogConsumer - Error occurred in partition processor for partitio