我有一台服务器需要保留所有用户的内存缓存。因此，假设列表不会很大 - 几十万个项目，我想使用带有键控消息的 Kafka 主题，其中键是 userId 以保持该列表的当前状态，并且管理应用程序将发送新的用户对象当事情发生变化时，转到那个话题。因此，当服务器启动时，它只需要从头开始读取该主题的所有内容并填充它的缓存。
填充阶段大约需要 20-30 秒，具体取决于与 Kafka 的连接，因此服务器不需要在线，直到它从主题中读取所有内容以拥有最新的缓存（主题中的所有消息都位于开始时刻被认为是最新的）。但我不知道如何确定我是否从 Kafka 流中读取所有内容以通知其他服务缓存已填充并且服务器可以启动服务器请求。我读过关于高水位线的文章，但没有看到它在 Java 消费者 API 中公开。
那么如何找出 Kafka 主题的最新偏移量以了解我的阅读器何时是最新的？
【问题讨论】：
标签： apache-kafka kafka-consumer-api
假设您使用的是高级消费者。
高水印在高级消费者中不可用。
**As you mentioned: all the messages in the topic at the moment of start is considered up-to-date**
当您的应用程序启动时，您可以使用 SimpleConsumer Api 执行以下操作：-
通过向 kafka 集群中的任何代理发出 TopicMetadataRequest 来查找主题中的分区数。
创建分区到 latestOffset 映射，其中键是分区，值是该分区中可用的 latestOffset。
Map
对于Topic中的每个分区p：
A.找到partition p的leader
B.向leader发送OffsetRequest
C.从 OffsetResponse 中获取最新的Offset
D.向 offsetMap 添加一个条目，其中键是分区 p，偏移量是 最新偏移量。
开始使用 高级消费者从 kafka 读取消息：
A.对于您从 KafkaStream 获得的每条消息：
AA. Get the partition && offset of the message
BB. if( offsetMap.get(partition)<=offset) stop Reading from this steam
希望这会有所帮助。
【讨论】：