spring kafka 在主题中寻找最新的可用消息

Question

我有一个 Kafka 消费者，它订阅了以下主题 MY_TOPIC、MY_UNINTERESTED_TOPIC。

在下面的场景中，我对第二个主题不感兴趣，但我不得不提及它，因为如果我使用 auto.offset.reset 之类的东西配置它，它可能会影响所有主题。

在MY_TOPIC 主题上，我发布了不同类型的消息：MESSAGE_TYPE_A 和MESSAGE_TYPE_B。两条消息都是BaseKafkaMessage（自定义类）的实例，具有不同的属性。

现在我有兴趣仅找到 MESSAGE_TYPE_A 类型的最新消息。我该怎么做？

真正的场景是这样的：我在同一个主题上发布两种类型的消息。其中一个用于在每个对此主题和该消息感兴趣的消费者中准备一个本地缓存。如果消费者停止，当它重新加载时，它必须使用最新的MESSAGE_TYPE_A 重新初始化其缓存。 MESSAGE_TYPE_B 应该被忽略。我不想在 Kafka 上向数据提供者发送通知以再次发布数据，因为所有订阅者都会有很多不必要的工作要做。

我怎样才能获得这个？这可能吗？

我找到了https://docs.spring.io/spring-kafka/reference/htmlsingle/#seek，但我不确定这是否是我正在寻找的，或者是否有其他方法可以做到这一点。

Answer 1

目前还不清楚这些消息采用什么格式，或者为什么您真的需要它们出现在同一个主题中。

例如，您可以use different Avro types。或者你必须try-catch 解析两个不同的字节数组（JSON 对象？）

或者您可以按类型对主题进行分区，这样一种类型的所有消息都按同一分区的顺序排列。

但是，没有进行索引查找以获取最近发送的消息的机制。要么从最新事件开始，然后收到下一条传入消息，要么从头开始，然后向上扫描，直到在下一个轮询循环中获得 0 条记录，这在理论上是“最近的”

类似 auto.offset.reset 的东西可能会影响所有主题

这只影响消费者感兴趣的主题，而不是全部。